RabbitMQ 学习笔记

AMQP协议模型

• Server：又称为Broker，接受客户端的链接，实现AMQP实体服务
  
• Connection：连接，应用程序与Broker的网络连接
  
• channel：网络信道，几乎所有的操作都在channel中进行，是消息读写的通道，可建立多个channel，每个channel代表一个会话任务
  
• Message：消息本体，由Properties和Body组成，Proerties对消息进行装饰（优先级、延迟等特征），body是消息体内容
  
• Virtual host：虚拟地址，进行逻辑各级，一个VirtualHost里面可以有若干个Exchange和Queue，同一个VirtualHost不能有相同的Exchange或Queue
  
• Exchange：交换机，接受消息，根据路由键转发消息到绑定的队列
  
• Binding：Exchange和Queue之间的虚拟链接 binding中可以报刊rounting key
  
• Rounting key ：消息发送的路由规则
  
• Queue： 消息队列，保存并转发给消费者
  AMQP 消息流转
  

命令行与管理台

1. rabbitmqctl stop_app           关闭应用
2. rabbitmqctl start_app          启动应用
3. rabbitmqctl status                节点状态
4. rabbitmqctl add_user username password        添加用户
5. rabbitmqctl list_users                                        打印用户列表
6. rabbitmqctl delete_user username 删除用户
7. rabbitmqctl clear_permissions -p vhostpath username  用户权限清除
8. rabbitmqctl add_vhost vhostpath 创建虚拟主机
9. rabbitmqctl list_vhosts 打印虚拟主机列表
10. rabbitmqctl list_queues 打印队列信息
11. rabbitmqctl -p vhostpath purge_queue blue 清除队列里的消息

复制代码

Exchange 交换机

Exchange：接受消息，并根据路由键转发消息所绑定的队列

Exchange 交换机属性

• Name：交换机名称
  
• Type：类型
  
  
  
  • direct 直连，发送到direct Exchange的消息被转发到RouteKey指定的Queue  exchange=doutingKey
    
  • Topic  发布\订阅，发送到Topic Exchange的消息会被转发到所有关心RouteKey中制定的Topic的Queue上
    
  • fanout、不处理路由键，简单的将队列板绑定到交换机上，发送到交换机上的消息都会被转发到该交换机绑定的队列上、fount是最快的
    
  • headers
    
  
  
• Durability：是否需要持久化
  
• Auto Delete：当最后一个绑定在Exchange上的队列删除后，自动删除Exchange
  
• Interal：当前exchange是否用于Rabbitmq内部使用，默认为FALSE
  
• Arguments：扩展参数，用户扩展AMQP协议参数
  

Binding-绑定

• Exchange和Exchange、queue之间绑定
  

消息可靠性投递 100%

生产端-可靠性投递

解决方案

• 消息落库、对消息状态达标
  
• 消息延迟投递、做二次确认，回掉检查
  

消费端-幂等性保证

解决方案：

• 数据库ID+指纹码
  
• 利用Redis原子性实现
  

Confirm确认消息

消费者收到消息会给生产者一个应答

1. channel.addConfirmListener

复制代码

Return 消息机制

RabbitMQ代码整合

Spring-AMQP

• RabbitAdmin -- Exchange\Queue\Binding 之间关系声明
  
• SpringAMQP 声明
  
• RabbitTemplate
  
• SimpleMessageListenerContainer
  
• MessageListenerAdapter
  
• MessageConverter
  RabbitAdmin
  
  1. package com.study.rabbitmq.config;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.Queue;
  6. import org.springframework.amqp.core.TopicExchange;
  7. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
  8. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  9. import org.springframework.amqp.rabbit.core.RabbitAdmin;
  10. import org.springframework.context.annotation.Bean;
  11. import org.springframework.stereotype.Component;
  13. @Component
  14. public class RabbitMQConfig {
  16.     @Bean
  17.     public ConnectionFactory connectionFactory(){
  18.         CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
  19.         connectionFactory.setAddresses("192.168.223.128:5672");
  20.         connectionFactory.setUsername("guest");
  21.         connectionFactory.setPassword("guest");
  22.         connectionFactory.setVirtualHost("/");
  23.         return connectionFactory;
  24.     }
  26.     @Bean
  27.     public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
  28.         RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
  29.         // setAutoStartup  必须设置为true 否则Spring不会加载RabbitAdmin
  30.         rabbitAdmin.setAutoStartup(true);
  31.         return rabbitAdmin;
  32.     }
  35.     /**
  36.      * 针对消费者配置
  37.      * 1. 设置交换机类型
  38.      * 2. 将队列绑定到交换机
  39.      FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
  40.      HeadersExchange ：通过添加属性key-value匹配
  41.      DirectExchange:按照routingkey分发到指定队列
  42.      TopicExchange:多关键字匹配
  43.      */
  44.     @Bean
  45.     public TopicExchange exchange001() {
  46.         return new TopicExchange("topic001", true, false);
  47.     }
  49.     @Bean
  50.     public Queue queue001() {
  51.         return new Queue("queue001", true); //队列持久
  52.     }
  54.     @Bean
  55.     public Binding binding001() {
  56.         return BindingBuilder.bind(queue001()).to(exchange001()).with("spring.*");
  57.     }
  58. }

  复制代码
  RabbitTemplate
  

1.    @Bean
2.     public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
3.         RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
4.         return rabbitTemplate;
5.     }
6.    
7.         @Test
8.     public void testSendMessage() throws Exception {
9.         //1 创建消息
10.         MessageProperties messageProperties = new MessageProperties();
11.         messageProperties.getHeaders().put("desc", "信息描述..");
12.         messageProperties.getHeaders().put("type", "自定义消息类型..");
13.         Message message = new Message("Hello RabbitMQ".getBytes(), messageProperties);
15.         rabbitTemplate.convertAndSend("topic001", "spring.amqp", message, new MessagePostProcessor() {
16.             @Override
17.             public Message postProcessMessage(Message message) throws AmqpException {
18.                 System.err.println("------添加额外的设置---------");
19.                 message.getMessageProperties().getHeaders().put("desc", "额外修改的信息描述");
20.                 message.getMessageProperties().getHeaders().put("attr", "额外新加的属性");
21.                 return message;
22.             }
23.         });
24.     }

复制代码

SimpleMessageListenerContainer

• SimpleMessageListenerContainer可以进行动态监听消息，设置事务、事务管理、事务属性、设置消费者数量，进行批量消费,设置消息确认模式、自动确认，是否重回队列、异常捕获header函数，设置消费者标签生成策略，是否独占模式，消费者属性，设置具体的监听器、消息转换器
  

1. @Bean
2.     public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){
3.         SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
4.         // 容器添加监听队列
5.         container.setQueues(queue001());
6.         // 设置监听数据量
7.         container.setConcurrentConsumers(1);
8.         container.setMaxConcurrentConsumers(5);
9.         // 是否重回队列 - FALSE
10.         container.setDefaultRequeueRejected(false);
11.         // 签收 - auto 自动签收
12.         container.setAcknowledgeMode(AcknowledgeMode.AUTO);
13.         //
14.         container.setExposeListenerChannel(true);
15.         // 设置客户端tag
16.         container.setConsumerTagStrategy(new ConsumerTagStrategy() {
17.             @Override
18.             public String createConsumerTag(String queue) {
19.                 return queue + "_" + UUID.randomUUID().toString();
20.             }
21.         });
22.         // 消息监听处理
23.         container.setMessageListener(new ChannelAwareMessageListener() {
24.             @Override
25.             public void onMessage(Message message, Channel channel) throws Exception {
26.                 System.out.println("---- 处理消费者"+new String(message.getBody()));
27.             }
28.         });
29.         return container;
30.     }

复制代码

消息监听MessageListenerAdapter 以及 消息封装处理 MessageConverter

1. // 消息坚定
2. MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
3. adapter.setDefaultListenerMethod("consumeMessage"); // 设置监听处理方法名
4. adapter.setMessageConverter(new TextMessageConverter()); // 设置消息封装类
5. container.setMessageListener(adapter);
7. // TextMessageConverter
8. public class TextMessageConverter implements MessageConverter {
9.     @Override
10.     public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
11.         return new Message(object.toString().getBytes(), messageProperties);
12.     }
14.     @Override
15.     public Object fromMessage(Message message) throws MessageConversionException {
16.         String contentType = message.getMessageProperties().getContentType();
17.         if(null != contentType && contentType.contains("text")) {
18.             return new String(message.getBody());
19.         }
20.         return message.getBody();
21.     }
22. }
23. // MessageDelegate
24. public class MessageDelegate {
25.     public void handleMessage(byte[] messageBody) {
26.         System.err.println("默认方法, 消息内容:" + new String(messageBody));
27.     }
29.     public void consumeMessage(byte[] messageBody) {
30.         System.err.println("字节数组方法, 消息内容:" + new String(messageBody));
31.     }
33.     public void consumeMessage(String messageBody) {
34.         System.err.println("字符串方法, 消息内容:" + messageBody);
35.     }
37.     public void method1(String messageBody) {
38.         System.err.println("method1 收到消息内容:" + new String(messageBody));
39.     }
41.     public void method2(String messageBody) {
42.         System.err.println("method2 收到消息内容:" + new String(messageBody));
43.     }
46.     public void consumeMessage(Map messageBody) {
47.         System.err.println("map方法, 消息内容:" + messageBody);
48.     }
51.     public void consumeMessage(Order order) {
52.         System.err.println("order对象, 消息内容, id: " + order.getId() +
53.                 ", name: " + order.getName() +
54.                 ", content: "+ order.getContent());
55.     }
57.     public void consumeMessage(File file) {
58.         System.err.println("文件对象 方法, 消息内容:" + file.getName());
59.     }
60. }

复制代码

SpringBoot整合RabbitMQ

1. // 生产者
2. @Component
3. public class RabbitSender {
5.     @Autowired
6.     private RabbitTemplate rabbitTemplate;
9.     //回调函数: confirm确认
10.     final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
11.         @Override
12.         public void confirm(CorrelationData correlationData, boolean ack, String cause) {
13.             System.err.println("correlationData: " + correlationData);
14.             System.err.println("ack: " + ack);
15.             if(!ack){
16.                 System.err.println("异常处理....");
17.             }
18.         }
19.     };
21.     //回调函数: return返回
22.     final RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
23.         @Override
24.         public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
25.                                     String exchange, String routingKey) {
26.             System.err.println("return exchange: " + exchange + ", routingKey: "
27.                     + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
28.         }
29.     };
32.     //发送消息方法调用: 构建Message消息
33.     public void send(Object message, Map<String, Object> properties) throws Exception {
34.         MessageHeaders mhs = new MessageHeaders(properties);
35.         Message msg = MessageBuilder.createMessage(message, mhs);
36.         rabbitTemplate.setConfirmCallback(confirmCallback);
37.         rabbitTemplate.setReturnCallback(returnCallback);
38.         //id + 时间戳 全局唯一
39.         CorrelationData correlationData = new CorrelationData("1234567890");
40.         rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);
41.     }
43.     //发送消息方法调用: 构建自定义对象消息
44.     public void sendOrder(Order order) throws Exception {
45.         rabbitTemplate.setConfirmCallback(confirmCallback);
46.         rabbitTemplate.setReturnCallback(returnCallback);
47.         //id + 时间戳 全局唯一
48.         CorrelationData correlationData = new CorrelationData("0987654321");
49.         rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);
50.     }
52. }

复制代码

1. // 消费者
2. @Component
3. public class RabbitReceiver {
5.     @RabbitListener(bindings = @QueueBinding(
6.             value = @Queue(value = "queue-1",
7.                     durable="true"),
8.             exchange = @Exchange(value = "exchange-1",
9.                     durable="true",
10.                     type= "topic",
11.                     ignoreDeclarationExceptions = "true"),
12.             key = "springboot.*"
13.     )
14.     )
15.     @RabbitHandler
16.     public void onMessage(Message message, Channel channel) throws Exception {
17.         System.err.println("--------------------------------------");
18.         System.err.println("消费端Payload: " + message.getPayload());
19.         Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
20.         //手工ACK
21.         channel.basicAck(deliveryTag, false);
22.     }
25.     /**
26.      *
27.      * @param order
28.      * @param channel
29.      * @param headers
30.      * @throws Exception
31.      */
32.     @RabbitListener(bindings = @QueueBinding(
33.             value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}",
34.                     durable="${spring.rabbitmq.listener.order.queue.durable}"),
35.             exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}",
36.                     durable="${spring.rabbitmq.listener.order.exchange.durable}",
37.                     type= "${spring.rabbitmq.listener.order.exchange.type}",
38.                     ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),
39.             key = "${spring.rabbitmq.listener.order.key}"
40.     )
41.     )
42.     @RabbitHandler
43.     public void onOrderMessage(@Payload Order order,
44.                                Channel channel,
45.                                @Headers Map<String, Object> headers) throws Exception {
46.         System.err.println("--------------------------------------");
47.         System.err.println("消费端order: " + order.getId());
48.         Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
49.         //手工ACK
50.         channel.basicAck(deliveryTag, false);
51.     }
52. }

复制代码

1. # 配置文件
2. spring:
3.   rabbitmq:
4.     addresses: 192.168.11.76:5672
5.     username: guest
6.     password: guest
7.     virtual-host: /
8.     connection-timeout: 15000
10.     publisher-confirms: true # 生产者exchange确认
11.     publisher-returns: true # 生产者queue确认
12.     template:
13.       mandatory: true
14.     listener:
15.       simple:
16.         acknowledge-mode: manual
17.         concurrency: 5
18.         max-concurrency: 10
19.       order: # 自定义消费队列
20.         queue:
21.           name: queue-2
22.           durable: true
23.         exchange:
24.           name: exchange-2
25.           durable: true
26.           type: topic
27.           ignoreDeclarationExceptions: true
28.         key: springboot.*

复制代码

<img alt="image-20240221125440948" loading="lazy">
SpringCloudStream 整合Rabbitmq

1. /**
2. * 生产者
3. * 这里的Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称。
4. * 通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息。
5. */
6. public interface Barista {
7.     String OUTPUT_CHANNEL = "output_channel";
9.     //注解@Output声明了它是一个输出类型的通道，名字是output_channel。这一名字与app1中通道名一致，表明注入了一个名字为output_channel的通道，类型是output，发布的主题名为mydest。
10.     @Output(Barista.OUTPUT_CHANNEL)
11.     MessageChannel logoutput();
13. }
16. @EnableBinding(Barista.class)
17. @Service
18. public class RabbitmqSender {
19.     @Autowired
20.     private Barista barista;
22.     // 发送消息
23.     public String sendMessage(Object message, Map<String, Object> properties) throws Exception {
24.         try{
25.             MessageHeaders mhs = new MessageHeaders(properties);
26.             Message msg = MessageBuilder.createMessage(message, mhs);
27.             boolean sendStatus = barista.logoutput().send(msg);
28.             System.err.println("--------------sending -------------------");
29.             System.out.println("发送数据：" + message + ",sendStatus: " + sendStatus);
30.         }catch (Exception e){
31.             System.err.println("-------------error-------------");
32.             e.printStackTrace();
33.             throw new RuntimeException(e.getMessage());
35.         }
36.         return null;
37.     }
38. }

复制代码

1. /**
2. * 消费者
3. * 这里的Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称。
4. * 通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息。
5. */
6. public interface ConsumerBarista {
7.     String INPUT_CHANNEL = "input_channel";
9.     //注解@Input声明了它是一个输入类型的通道，名字是Barista.INPUT_CHANNEL，也就是position3的input_channel。这一名字与上述配置app2的配置文件中position1应该一致，表明注入了一个名字叫做input_channel的通道，它的类型是input，订阅的主题是position2处声明的mydest这个主题
10.     @Input(ConsumerBarista.INPUT_CHANNEL)
11.     SubscribableChannel loginput();
13. }
15. @EnableBinding(ConsumerBarista.class)
16. @Service
17. public class RabbitmqReceiver {
19.     @StreamListener(ConsumerBarista.INPUT_CHANNEL)
20.     public void receiver(Message message) throws Exception {
21.         Channel channel = (com.rabbitmq.client.Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
22.         Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
23.         System.out.println("Input Stream 1 接受数据：" + message);
24.         System.out.println("消费完毕------------");
25.         channel.basicAck(deliveryTag, false);
26.     }
27. }

复制代码

1. # 服务配置
2. server:
3.   port: 8888
4.   context-path: /rabbitmq
5. # spring应用配置
6. spring:
7.   application:
8.     name: rabbitmq-cloud-stream
9.   # SpringCloud 服务配置
10.   cloud:
11.     stream:
12.       binders:
13.         output_channel:
14.           destination: exchange-3
15.           group: queue-3
16.           binder: rabbit_cluster
17.         input_channel:
18.           destination: exchange-3
19.           group: queue-3
20.           binder: rabbit_cluster
21.           consumer:
22.             concurrency: 1
23.       bindings:
24.         rabbit_cluster:
25.           type: rabbit
26.           environment:
27.             spring:
28.               rabbitmq:
29.                 addresses: 192.168.11.76:5672
30.                 username: guest
31.                 password: guest
32.                 virtual-host: /
33.         input_channel:
34.           consumer:
35.             requeue-rejected: false
36.             acknowledge-mode: MANUAL
37.             recovery-interval: 3000
38.             durable-subscription: true
39.             max-concurrency: 5

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！