mirror of
https://gitee.com/wangkang/udf.git
synced 2026-10-29 07:03:47 +00:00
更新
This commit is contained in:
@@ -0,0 +1,174 @@
|
||||
# spring boot / cloud (九) 使用rabbitmq消息中间件
|
||||
|
||||
##前言
|
||||
|
||||
|
||||
###rabbitmq介绍:
|
||||
|
||||
|
||||
>RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统。它可以用于大型软件系统各个模块之间的高效通信,支持高并发,支持可扩展。
|
||||
|
||||
|
||||
###amqp介绍:
|
||||
|
||||
|
||||
>即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同的开发语言等条件的限制。Erlang中的实现有 RabbitMQ等。
|
||||
|
||||
|
||||
##思路
|
||||
|
||||
|
||||
基于spring boot的特性连接rabbitmq,并作出如下样例:
|
||||
|
||||
|
||||
- 配置
|
||||
|
||||
|
||||
- 发布方样例
|
||||
|
||||
|
||||
- 消费方样例
|
||||
|
||||
|
||||
##实现
|
||||
|
||||
|
||||
###1.配置
|
||||
|
||||
引入maven依赖
|
||||
|
||||
``` xml
|
||||
<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传输)
|
||||
|
||||
``` java
|
||||
@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);
|
||||
}
|
||||
|
||||
}
|
||||
```
|
||||
|
||||
编写配置文件
|
||||
|
||||
``` properties
|
||||
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 #消费方是否重试
|
||||
```
|
||||
|
||||
定义公共的消息类
|
||||
|
||||
``` java
|
||||
public class RabbitmqMessage<T> implements Serializable {
|
||||
private static final long serialVersionUID = 1L;
|
||||
//消息ID
|
||||
private String id;
|
||||
....其他自定义
|
||||
}
|
||||
```
|
||||
|
||||
|
||||
###2.发布方样例
|
||||
|
||||
创建Exchange,这里使用的是DirectExchange,exchange主要是定义路由规则的,还有其他不同的路由规则实现,如:TopicExchange,他们都继承至AbstractExchange
|
||||
|
||||
``` java
|
||||
@Bean
|
||||
public DirectExchange testExchange() {
|
||||
return new DirectExchange("test_exchange");
|
||||
}
|
||||
```
|
||||
|
||||
使用AmqpTemplate发送异步消息(RoutingKey则是指定消息的路由键,不同的路由键可被不同的消费方消费)
|
||||
|
||||
``` java
|
||||
@Autowired
|
||||
private AmqpTemplate amqpTemplate;
|
||||
|
||||
//然后调用发送方法发送消息
|
||||
this.amqpTemplate.convertAndSend("test_exchange", "testRoutingKey", new RabbitmqMessage<String>("test"));
|
||||
```
|
||||
|
||||
###3.消费方样例
|
||||
|
||||
创建消费队列,死信队列,以及与exchange的绑定关系
|
||||
|
||||
``` java
|
||||
//消费队列
|
||||
@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);
|
||||
}
|
||||
```
|
||||
|
||||
消费消息
|
||||
|
||||
``` java
|
||||
@RabbitListener(queues = "test_consume")
|
||||
public void process(Message<String> message) {
|
||||
log.info(message);
|
||||
}
|
||||
```
|
||||
|
||||
|
||||
##结束
|
||||
|
||||
以上演示了rabbitmq在spring boot中的配置,以及发送方和消费方的样例,在后续的章节中,会找机会介绍rabbitmq的搭建以及配置.
|
||||
|
||||
---------
|
||||
|
||||
想获得最快更新,请关注公众号
|
||||
|
||||

|
||||
|
||||
Reference in New Issue
Block a user