spring boot / cloud (九) 使用rabbitmq消息中间件

spring boot / cloud (九) 使用rabbitmq消息中间件

前言

rabbitmq介绍:

amqp介绍:

思路

基于spring boot的特性连接rabbitmq,并作出如下样例:

配置

发布方样例

消费方样例

实现

1.配置

引入maven依赖

<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

编写config配置类(默认情况下是不用做任何配置的,这里有配置是因为,它默认是用的二进制做的消息传输,这里的配置是改为json传输)

@Configuration
public class RabbitMqConfig {

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactoryPlus(
SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory,
Jackson2JsonMessageConverter jackson2JsonMessageConverter) {
rabbitListenerContainerFactory.setMessageConverter(jackson2JsonMessageConverter);
return rabbitListenerContainerFactory;
}

@Bean
public Jackson2JsonMessageConverter jackson2JsonMessageConverter(ObjectMapper xssObjectMapper) {
return new Jackson2JsonMessageConverter(xssObjectMapper);
}

}

编写配置文件

spring.rabbitmq.host=192.168.134.100
spring.rabbitmq.port=5672
spring.rabbitmq.username=dev_udf-sample
spring.rabbitmq.password=1qazxsw2
spring.rabbitmq.virtual-host=/dev_udf-sample
spring.rabbitmq.template.retry.enabled=true #发送方是否重试
spring.rabbitmq.listener.retry.enabled=true #消费方是否重试

定义公共的消息类

public class RabbitmqMessage<T> implements Serializable {
private static final long serialVersionUID = 1L;
//消息ID
private String id;
....其他自定义
}

2.发布方样例

创建Exchange,这里使用的是DirectExchange,exchange主要是定义路由规则的,还有其他不同的路由规则实现,如:TopicExchange,他们都继承至AbstractExchange

@Bean
public DirectExchange testExchange() {
return new DirectExchange("test_exchange");
}

使用AmqpTemplate发送异步消息(RoutingKey则是指定消息的路由键,不同的路由键可被不同的消费方消费)

@Autowired
private AmqpTemplate amqpTemplate;

//然后调用发送方法发送消息
this.amqpTemplate.convertAndSend("test_exchange", "testRoutingKey", new RabbitmqMessage<String>("test"));

3.消费方样例

创建消费队列,死信队列,以及与exchange的绑定关系

//消费队列
@Bean
public Queue testConsume() {
//死信exchange与上面的定义方式一样
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange","test_exchange_dlx");
args.put("x-dead-letter-routing-key","testRoutingKey_dlx");
return new Queue("test_consume", true, false, false, args);
}

//死信消费队列
@Bean
public Queue testConsumeDlx() {
return new Queue("test_consume_dlx");
}

//消费队列绑定
@Bean
public Binding testConsumeBinding() {
return new Binding("test_consume", DestinationType.QUEUE,
"test_exchange","testRoutingKey", null);
}

//死信消费队列绑定
@Bean
public Binding testConsumeDlxBinding() {
return new Binding("test_consume_dlx", DestinationType.QUEUE,
"test_exchange_dlx","testRoutingKey_dlx", null);
}

消费消息

@RabbitListener(queues = "test_consume")
public void process(Message<String> message) {
log.info(message);
}

代码仓库 (博客配套代码)

udf-starter : 基础项目,脚手架,框架

udf-sample : 集成样例

结束

以上演示了rabbitmq在spring boot中的配置,以及发送方和消费方的样例,在后续的章节中,会找机会介绍rabbitmq的搭建以及配置.

想获得最快更新,请关注公众号