• RabbitMQ基础

    目录

    MQ 

    MQ概述

    MQ 的优势 

    1.应用解耦

    2.异步提速

    3.削峰填谷

    MQ 的劣势

    1.系统可用性降低

    2.系统复杂度提高

    3.一致性问题

    使用 MQ 需要满足什么条件呢？

     RabbitMQ 简介

    ​编辑RabbitMQ 中的相关概念

    RabbitMQ 提供了 6 种工作模式

    JMS

    java实现Rabbitmq

    依赖导入和基础配置

    简单模式

    Work queues 工作队列模式 

    Pub/Sub 订阅模式

    Routing 路由模式

    Topics 通配符模式

     Spring整合Rabbitmq

    供应者

    消费者

     SpringBoot整合Rabbitmq

     

    MQ 

            目前业界有很多的 MQ 产品，如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等，也有直接使用 Redis 充当消息队列的案例，而这些消息队列产品，各有侧重，在实际选型时，需要结合自身需 求及 MQ 产品特征，综合考虑。

    MQ概述

            MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进 行通信。

    MQ 的优势 

    1.应用解耦

    使用 MQ 使得应用间解耦，提升容错性和可维护性。

    2.异步提速

    一个下单操作耗时：20 + 300 + 300 + 300 = 920ms 用户点击完下单按钮后，需要等待920ms才能得到下单响应，太慢了

    用户点击完下单按钮后，只需等待25ms就能得到下单响应 (20 + 5 = 25ms)。

    提升用户体验和系统吞吐量（单位时间内处理请求的数目）。

    3.削峰填谷

            MQ 之后，限制消费消息的速度为1000，这样一来，高峰期产生的数据势必会被积压在 MQ 中，高峰 就被“削”掉了，但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000，直 到消费完积压的消息，这就叫做“填谷”。

    MQ 的劣势

    1.系统可用性降低

            系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

    2.系统复杂度提高

            MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何 保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

    3.一致性问题

            A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理 失败。如何保证消息数据处理的一致性？

    使用 MQ 需要满足什么条件呢？

    • 生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明 明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。
    • 容许短暂的不一致性。
    • 确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。

     RabbitMQ 简介

            AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议 的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中 间件不同产品，不同的开发语言等条件的限制。2006年，AMQP 规范发布。类比HTTP。

    RabbitMQ 中的相关概念

    ⚫ Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker

    ⚫ Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多 个vhost，每个用户在自己的 vhost 创建 exchange／queue

    ⚫ Connection：publisher／consumer 和 broker 之间的 TCP 连接

    ⚫ Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection 的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线 程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和 message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销

    ⚫ Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到 queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)

    ⚫ Queue：消息最终被送到这里等待 consumer 取走

    ⚫ Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存 到 exchange 中的查询表中，用于 message 的分发依据

    RabbitMQ 提供了 6 种工作模式

            简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing 路由模式、Topics 主题模式、RPC 远程调用模式（远程调用，不太算 MQ；暂不作介绍）。 官网对应模式介绍：https://www.rabbitmq.com/getstarted.html

    JMS

            JMS 即 Java 消息服务（JavaMessage Service）应用程序接口，是一个 Java 平台中关于面向消息中间件的API

            JMS 是 JavaEE 规范中的一种，API 规范接口，类比JDBC


            很多消息中间件都实现了JMS规范，例如：ActiveMQ。RabbitMQ 官方没有提供 JMS 的实现包，但是开源社区有

    java实现Rabbitmq

    依赖导入和基础配置

    1.    <dependencies>
    2.         <!--rabbitmq java 客户端-->
    3.         <dependency>
    4.             <groupId>com.rabbitmq</groupId>
    5.             <artifactId>amqp-client</artifactId>
    6.             <version>5.6.0</version>
    7.         </dependency>
    8.     </dependencies>
    9.  
    10.  
    11.     <build>
    12.         <plugins>
    13.             <plugin>
    14.                 <groupId>org.apache.maven.plugins</groupId>
    15.                 <artifactId>maven-compiler-plugin</artifactId>
    16.                 <version>3.8.0</version>
    17.                 <configuration>
    18.                     <source>1.8</source>
    19.                     <target>1.8</target>
    20.                 </configuration>
    21.             </plugin>
    22.         </plugins>
    23.     </build>

    基础配置信息 

    1.         //1.创建连接工厂
    2.         ConnectionFactory factory = new ConnectionFactory();
    3.         //2. 设置参数
    4.         factory.setHost("121.37.118.193");//ip  默认值 localhost
    5.         factory.setPort(5672); //端口  默认值 5672
    6.         factory.setVirtualHost("dobby");//虚拟机 默认值/
    7.         factory.setUsername("dobby");//用户名 默认 guest
    8.         factory.setPassword("123456");//密码 默认值 guest
    9.         //3. 创建连接 Connection
    10.         Connection connection = factory.newConnection();
    11.         //4. 创建Channel
    12.         Channel channel = connection.createChannel();

    简单模式

    • P：生产者，也就是要发送消息的程序
    • C：消费者：消息的接收者，会一直等待消息到来
    • queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息
    1.         //创建队列Queue
    2.         /*
    3.         queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
    4.         参数：
    5.             1. queue：队列名称
    6.             2. durable:是否持久化，当mq重启之后，还在
    7.             3. exclusive：
    8.                 * 是否独占。只能有一个消费者监听这队列
    9.                 * 当Connection关闭时，是否删除队列
    10.                 *
    11.             4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
    12.             5. arguments：参数。
    14.          */
    15.         //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
    16.         channel.queueDeclare("hello_world",true,false,false,null);
    17.         /*
    18.         basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
    19.         参数：
    20.             1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
    21.             2. routingKey：路由名称
    22.             3. props：配置信息
    23.             4. body：发送消息数据
    25.          */
    26.  
    27.         String body = "hello rabbitmq~~~";
    28.  
    29.         //发送消息
    30.         channel.basicPublish("","hello_world",null,body.getBytes());

    Work queues 工作队列模式 

    • Work Queues：与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
    • 在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。

    与简单模式供应端代码一致，只不过对应多个消费端，以下消费端代码示例

    1.         String queue1Name = "test_fanout_queue1";
    2.  
    3.         // 接收消息
    4.         Consumer consumer = new DefaultConsumer(channel){
    5.             /*
    6.                 回调方法，当收到消息后，会自动执行该方法
    8.                 1. consumerTag：标识
    9.                 2. envelope：获取一些信息，交换机，路由key...
    10.                 3. properties:配置信息
    11.                 4. body：数据
    13.              */
    14.             @Override
    15.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    16.               /*  System.out.println("consumerTag："+consumerTag);
    17.                 System.out.println("Exchange："+envelope.getExchange());
    18.                 System.out.println("RoutingKey："+envelope.getRoutingKey());
    19.                 System.out.println("properties："+properties);*/
    20.                 System.out.println("body："+new String(body));
    21.                 System.out.println("将日志信息保存数据库.....");
    22.             }
    23.         };
    24.         /*
    25.         basicConsume(String queue, boolean autoAck, Consumer callback)
    26.         参数：
    27.             1. queue：队列名称
    28.             2. autoAck：是否自动确认
    29.             3. callback：回调对象
    31.          */
    32.         channel.basicConsume(queue2Name,true,consumer);

    Pub/Sub 订阅模式

            在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：

    •  P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
    •  C：消费者，消息的接收者，会一直等待消息到来
    •  Queue：消息队列，接收消息、缓存消息
    •  Exchange：交换机（X）。

            一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：

    •  Fanout：广播，将消息交给所有绑定到交换机的队列
    •  Direct：定向，把消息交给符合指定routing key 的队列
    •  Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

            Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！ 

            发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队 列绑 定到默认的交换机

    1.         String exchangeName = "test_fanout";
    2.         //创建交换机
    3.        /*exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map arguments)
    4.        参数：
    5.         1. exchange:交换机名称
    6.         2. type:交换机类型
    7.             DIRECT("direct"),：定向
    8.             FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
    9.             TOPIC("topic"),通配符的方式
    10.             HEADERS("headers");参数匹配
    12.         3. durable:是否持久化
    13.         4. autoDelete:自动删除
    14.         5. internal：内部使用。 一般false
    15.         6. arguments：参数
    16.         */
    17.         channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
    18.         //创建队列
    19.         String queue1Name = "test_fanout_queue1";
    20.         String queue2Name = "test_fanout_queue2";
    21.         channel.queueDeclare(queue1Name,true,false,false,null);
    22.         channel.queueDeclare(queue2Name,true,false,false,null);
    23.         //绑定队列和交换机
    24.         /*
    25.         queueBind(String queue, String exchange, String routingKey)
    26.         参数：
    27.             1. queue：队列名称
    28.             2. exchange：交换机名称
    29.             3. routingKey：路由键，绑定规则
    30.                 如果交换机的类型为fanout ，routingKey设置为""
    31.          */
    32.         channel.queueBind(queue1Name,exchangeName,"");
    33.         channel.queueBind(queue2Name,exchangeName,"");
    34.  
    35.         String body = "日志信息：张三调用了findAll方法...日志级别：info...";
    36.         //发送消息
    37.         channel.basicPublish(exchangeName,"",null,body.getBytes());
    38.  
    39.         //释放资源
    40.         channel.close();
    41.         connection.close();

    Routing 路由模式


            Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的Routingkey 与消息的 Routing key 完全一致，才会接收到消息

    • P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key
    • X：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列
    • C1：消费者，其所在队列指定了需要 routing key 为 error 的消息
    • C2：消费者，其所在队列指定了需要 routing key 为 info、error、warning 的消息

            Routing 模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合 routing key 的队列。 

    1.         //创建交换机
    2.         channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);
    3.         //创建队列
    4.         String queue1Name = "test_direct_queue1";
    5.         String queue2Name = "test_direct_queue2";
    6.  
    7.         channel.queueDeclare(queue1Name,true,false,false,null);
    8.         channel.queueDeclare(queue2Name,true,false,false,null);
    9.         //绑定队列和交换机
    10.         //队列1绑定 error
    11.         channel.queueBind(queue1Name,exchangeName,"error");
    12.         //队列2绑定 info  error  warning
    13.         channel.queueBind(queue2Name,exchangeName,"info");
    14.         channel.queueBind(queue2Name,exchangeName,"error");
    15.         channel.queueBind(queue2Name,exchangeName,"warning");
    16.  
    17.         String body = "日志信息：张三调用了delete方法...出错误了。。。日志级别：error...";
    18.         //8. 发送消息
    19.         channel.basicPublish(exchangeName,"warning",null,body.getBytes());

    Topics 通配符模式

    1. 红色 Queue：绑定的是 usa.# ，因此凡是以 usa. 开头的 routing key 都会被匹配到
    2. 黄色 Queue：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配
    3. 通配符规则：# 匹配一个或多个词，* 匹配不多不少恰好1个词，例如：item.# 能够匹配 item.insert.abc 或者 item.insert，item.* 只能匹配 item.insert
    1.          //创建交换机
    2.         channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);
    3.         ...
    4.          // routing key  系统的名称.日志的级别。
    5.         //=需求： 所有error级别的日志存入数据库，所有order系统的日志存入数据库
    6.         channel.queueBind(queue1Name,exchangeName,"#.error");
    7.         channel.queueBind(queue1Name,exchangeName,"order.*");
    8.         channel.queueBind(queue2Name,exchangeName,"*.*");
    9.  
    10.         String body = "日志信息：张三调用了findAll方法...日志级别：info...";
    11.         //8. 发送消息
    12.         channel.basicPublish(exchangeName,"goods.error",null,body.getBytes());

     Spring整合Rabbitmq

    供应者

    rabbitmq.properties

    1. rabbitmq.host=121.37.118.193
    2. rabbitmq.port=5672
    3. rabbitmq.username=dobby
    4. rabbitmq.password=123456
    5. rabbitmq.virtual-host=dobby

    Spring-rabbitmq-prodicer.xml

    1. "1.0" encoding="UTF-8"?>
    2. <beans xmlns="http://www.springframework.org/schema/beans"
    3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.        xmlns:context="http://www.springframework.org/schema/context"
    5.        xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    6.        xsi:schemaLocation="http://www.springframework.org/schema/beans
    7.        http://www.springframework.org/schema/beans/spring-beans.xsd
    8.        http://www.springframework.org/schema/context
    9.        https://www.springframework.org/schema/context/spring-context.xsd
    10.        http://www.springframework.org/schema/rabbit
    11.        http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    12.     
    13.     <context:property-placeholder location="classpath:rabbitmq.properties"/>
    14.  
    15.     
    16.     <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
    17.                                port="${rabbitmq.port}"
    18.                                username="${rabbitmq.username}"
    19.                                password="${rabbitmq.password}"
    20.                                virtual-host="${rabbitmq.virtual-host}"/>
    21.     
    22.     <rabbit:admin connection-factory="connectionFactory"/>
    23.  
    24.     
    25.     
    26.  
    27.     <rabbit:queue id="spring_queue" name="spring_queue"    auto-declare="true"/>
    28.  
    29.     
    30.     
    31.     <rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" auto-declare="true"/>
    32.  
    33.     
    34.     <rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" auto-declare="true"/>
    35.  
    36.     
    37.     <rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange"  auto-declare="true">
    38.         <rabbit:bindings>
    39.             <rabbit:binding  queue="spring_fanout_queue_1"  />
    40.             <rabbit:binding queue="spring_fanout_queue_2"/>
    41.         rabbit:bindings>
    42.     rabbit:fanout-exchange>
    43.  
    44.     
    45.  
    46.     
    47.     
    48.     <rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star"  auto-declare="true"/>
    49.     
    50.     <rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/>
    51.     
    52.     <rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/>
    53.  
    54.     <rabbit:topic-exchange id="spring_topic_exchange"  name="spring_topic_exchange" auto-declare="true">
    55.         <rabbit:bindings>
    56.             <rabbit:binding pattern="heima.*"  queue="spring_topic_queue_star"/>
    57.             <rabbit:binding pattern="heima.#" queue="spring_topic_queue_well"/>
    58.             <rabbit:binding pattern="itcast.#" queue="spring_topic_queue_well2"/>
    59.         rabbit:bindings>
    60.     rabbit:topic-exchange>
    61.  
    62.     
    63.     <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
    64. beans>

    测试代码： 

    1.  
    2. @RunWith(SpringJUnit4ClassRunner.class)
    3. @ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
    4. public class ProducerTest {
    5.  
    6.     //1.注入 RabbitTemplate
    7.     @Autowired
    8.     private RabbitTemplate rabbitTemplate;
    9.  
    10.  
    11.     @Test
    12.     public void testHelloWorld(){
    13.         //2.发送消息
    14.  
    15.         rabbitTemplate.convertAndSend("spring_queue","hello world spring....");
    16.     }
    17.  
    18.  
    19.     /**
    20.      * 发送fanout消息
    21.      */
    22.     @Test
    23.     public void testFanout(){
    24.         //2.发送消息
    25.  
    26.         rabbitTemplate.convertAndSend("spring_fanout_exchange","","spring fanout....");
    27.     }
    28.  
    29.  
    30.     /**
    31.      * 发送topic消息
    32.      */
    33.     @Test
    34.     public void testTopics(){
    35.         //2.发送消息
    36.  
    37.         rabbitTemplate.convertAndSend("spring_topic_exchange","heima.hehe.haha","spring topic....");
    38.     }
    39. }
    SpringQueueListener
    1. import org.springframework.amqp.core.Message;
    2. import org.springframework.amqp.core.MessageListener;
    3.  
    4. public class SpringQueueListener implements MessageListener {
    5.     @Override
    6.     public void onMessage(Message message) {
    7.         //打印消息
    8.         System.out.println(new String(message.getBody()));
    9.     }
    10. }

    消费者

    Spring-rabbitmq-consumer.xml

    1. "1.0" encoding="UTF-8"?>
    2. <beans xmlns="http://www.springframework.org/schema/beans"
    3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.        xmlns:context="http://www.springframework.org/schema/context"
    5.        xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    6.        xsi:schemaLocation="http://www.springframework.org/schema/beans
    7.        http://www.springframework.org/schema/beans/spring-beans.xsd
    8.        http://www.springframework.org/schema/context
    9.        https://www.springframework.org/schema/context/spring-context.xsd
    10.        http://www.springframework.org/schema/rabbit
    11.        http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    12.     
    13.     <context:property-placeholder location="classpath:rabbitmq.properties"/>
    14.  
    15.     
    16.     <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
    17.                                port="${rabbitmq.port}"
    18.                                username="${rabbitmq.username}"
    19.                                password="${rabbitmq.password}"
    20.                                virtual-host="${rabbitmq.virtual-host}"/>
    21.  
    22.     <bean id="springQueueListener" class="com.itheima.rabbitmq.listener.SpringQueueListener"/>
    23.     
    25.     
    26.     <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true">
    27.         <rabbit:listener ref="springQueueListener" queue-names="spring_queue"/>
    28.        
    29.     rabbit:listener-container>
    30. beans>

     SpringBoot整合Rabbitmq

    1.     
    2.     <parent>
    3.         <groupId>org.springframework.bootgroupId>
    4.         <artifactId>spring-boot-starter-parentartifactId>
    5.         <version>2.1.4.RELEASEversion>
    6.     parent>
    7.  
    8.     <dependencies>
    9.         
    10.         <dependency>
    11.             <groupId>org.springframework.bootgroupId>
    12.             <artifactId>spring-boot-starter-amqpartifactId>
    13.         dependency>
    14.         <dependency>
    15.             <groupId>org.springframework.bootgroupId>
    16.             <artifactId>spring-boot-starter-testartifactId>
    17.         dependency>
    18.     dependencies>
    1. @Configuration
    2. public class RabbitMQConfig {
    3.  
    4.     public static final String EXCHANGE_NAME = "boot_topic_exchange";
    5.     public static final String QUEUE_NAME = "boot_queue";
    6.  
    7.     //1.交换机
    8.     @Bean("bootExchange")
    9.     public Exchange bootExchange(){
    10.         return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    11.     }
    12.  
    13.  
    14.     //2.Queue 队列
    15.     @Bean("bootQueue")
    16.     public Queue bootQueue(){
    17.         return QueueBuilder.durable(QUEUE_NAME).build();
    18.     }
    19.  
    20.     //3. 队列和交互机绑定关系 Binding
    21.     /*
    22.         1. 知道哪个队列
    23.         2. 知道哪个交换机
    24.         3. routing key
    25.      */
    26.     @Bean
    27.     public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange){
    28.         return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
    29.     }
    30.  
    31.  
    32. }
    1. # 配置RabbitMQ的基本信息  ip 端口 username  password..
    2. spring:
    3.   rabbitmq:
    4.     host: 172.16.98.133 # ip
    5.     port: 5672
    6.     username: guest
    7.     password: guest
    8.     virtual-host: /
    1. public class ProducerTest {
    2.  
    3.     //1.注入RabbitTemplate
    4.     @Autowired
    5.     private RabbitTemplate rabbitTemplate;
    6.  
    7.     @Test
    8.     public void testSend(){
    9.  
    10.         rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"boot.haha","boot mq hello~~~");
    11.     }
    12. }

    消费者：

    RabbimtMQListener
    1. @Component
    2. public class RabbimtMQListener {
    3.  
    4.     @RabbitListener(queues = "boot_queue")
    5.     public void ListenerQueue(Message message){
    6.         //System.out.println(message);
    7.         System.out.println(new String(message.getBody()));
    8.     }
    9.  
    10. }

  • 相关阅读:
    Electron + vue搭建项目
    POJ1007：DNA排序
    windows环境elasticsearch设置登录用户名、密码
    算力服务网络：一场多元融合的系统革命
    c语言 4.0
    MySQL 枚举类型如何定义比较好 tinyint?enum?varchar?
    Qt-OpenCV学习笔记--最小包覆矩形--minAreaRect()
    ros自定义消息包无法编译生成.h文件的问题解决
    vscode 源代码不能自动stage change
    android获取RAM、CPU频率、系统版本、CPU核数
  • 原文地址：https://blog.csdn.net/qq_45412648/article/details/132657010