devbox
酷酷是懒虫
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Rabbitmq_rabbitmq默认账号密码

作者：酷酷是懒虫 | 2024-06-29 16:39:30

赞

踩

rabbitmq默认账号密码

认识Rabbitmq

一、什么是Rabbitmq

RabbitMQ简称MQ是一套实现了高级消息队列协议的开源消息代理软件，简单来说就是一个消息中间件。是一种程序对程序的通信方法，其服务器也是以高性能、健壮以及可伸缩性出名的Erlang语言编写而成。

二、 Rabbitmq有什么作用

abbitMQ简单来说就是一个消息队列中间件，用来保存消息和传递消息的一个容器。在此过程中充当一个中间人的作用 而队列的主要目的就是提供正确的路由来保证消息的传递；如果发送消息时消费者不可用的话，默认情况下该消息将会一直被存储在队列中，直到消费者消费为止 那么同时呢，如果设置了消息存活的时间，即消息的有效期。在此有效期间消息如果还没有被消费的话，那么该消息就会变成死信，由死信交换机接收。而绑定死信交换机的队列则称为死信队列

三、RabbitMQ的常见作用

RabbitMQ的常见作用有三种，分别是服务间解耦、实现异步通信、流量削峰。

主要实现了消费者和生产者之间的解耦，发送异步消息，高并发访问解决流量削锋等问题。实现高性能，高可用，可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。

四、RabbitMQ的应用场景

场景一：用户订单，库存处理。【服务间解耦】

使用MQ前：系统正常时，用户下单，订单系统调用库存系统进行删减操作，操作成功，将成返回消息，提醒下单成功。系统异常时，库存系统将无法访问，导致订单删减操作无法执行，最终导致下单失败。

 

使用MQ后：订单系统和库存系统之间不在互相影响，独立运行，达到了应用解耦的目的。订单系统只需要将下单消息写入MQ,就可以直接执行下一步操作。这时即使库存系统出现异常也不会影响订单系统的操作，且下单的库存删减记录，将会被永久保存到MQ中，直到库存系统恢复正常，从MQ中订阅下单消息，进行消费成功为止。

 

场景二：用户注册，发送手机短信，邮件。【实现异步通信】

使用MQ前：整个操作流程，全部在主线程完成。点击用户注册 --》 入库添加用户 --》发送邮件 --》发送短信。每一步都需要等待上一步完成后才能执行。且每一步操作的响应时间不固定，如果请求过多，会导致主线程请求耗时很长，响应慢，甚至会导致死机的情况出现，严重影响了用户的体验。

使用MQ后：我们在大量用户进行秒杀请求时，将那个巨大的流量请求拒在系统业务处理的上层，并将其转移至MQ中，而不是直接涌入我们的接口。在这里MQ消息队列起到了缓存作用。

 

场景三：商品秒杀和抢购。【流量削峰】

流量削峰是消息队列中常用的场景 一般在秒杀或团购活动中使用广泛。

使用MQ前：对于秒杀、抢购活动，用户访问所产生的流量会很大，甚至会在同一时间段出现上万上亿条请求，这股瞬间的流量暴涨，我们的应用系统配置是无法承受的，会导致系统直接崩溃死机。

例如：A系统平时每秒请求100个，系统稳定运行； 但是晚上8点有秒杀活动 ，每秒并发增至1万条 ，系统最大处理每秒1000条 于是导致系统崩溃。

 

使用MQ后：我们在大量用户进行秒杀请求时，将那个巨大的流量请求拒在系统业务处理的上层，并将其转移至MQ中，而不是直接涌入我们的接口。在这里MQ消息队列起到了缓存作用。

例如：100万用户在高峰期，每秒请求5000个，将这5000个请求写入MQ系统每秒只能处理2000请求，因为MySQL只能处理2000个请求 ; 系统每秒拉取2000个请求 不超过自己的处理能力即可。

 

下载安装

Centos7 安装rabbitmq详细教程_江東-H的博客-CSDN博客

1、下载erlang

erlang官网http://www.erlang.org/downloads

准备环境：

yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel

设定安装规则：
./configure --prefix=/usr/local/erlang --with-ssl --enable-threads --enable-smp-support --enable-kernel-poll --enable-hipe --without-javac

安装：

make && make install

配置环境变量：

vim /etc/profile

export PATH=/root/RabbitMQ/erlang/otp_src_21.1/bin:${PATH}
source /etc/profile    #使环境变量生效

检验是否安装成功：

erl

退出用 halt()

2、下载rabbitmq

设置环境变量：export PATH=$PATH:/usr/local/rabbitmq_server-3.7.8/sbin

添加web管理插件命令：rabbitmq-plugins enable rabbitmq_management  启动端口15672

启动rabbitmq

安装目录下 /sbin       rabbitmq-server

登录rabbitmq    初始账号密码都是guest

3、初始化登录报：User can only log in via localhost

原因：

rabbitmq从3.3.0开始禁止使用guest/guest权限通过除localhost外的访问

解决问题：

找到安装目录 /ebin/rabbit.app

将：{loopback_users, [<<”guest”>>]}
改为：{loopback_users, []}

mq可视化界面

1.默认会提供一个默认用户guest，密码也是guest，线上环境需要创建一个新用户，并把guest用户删除。

2.首先切换到Admin标签页，可以查看或添加用户，添加用户时，可指定Tags，相当于角色，会拥有对应的权限：

1、Exhanges 功能

Durable 选择是否持久化

  • Durable 持久化
  • Transient 不持久化

Type-选择交换机类型

  • topic 主机模式
  • fanout 工作者模式
  • direct 路由模式

Auto Delete 是否自动删除

2、交换机详细操作

3、队列

切换到“Queues”标签，可以查看队列信息，点击队列名称，可查看队列所有状态的消息数量和大小等统计信息：

4、队列详细操作

一、常用的mq工具：

  • Rabbitmq
  • Activemq
  • Rocketmq
  • Kafka
  • Tubemq

二、mq的作用：

RabbitMQ是一种消息中间件，用于处理来自客户端的异步消息。服务端将要发送的消息放入到队列池中。接收端可以根据RabbitMQ配置的转发机制接收服务端发来的消息。RabbitMQ依据指定的转发规则进行消息的转发、缓冲和持久化操作，主要用在多服务器间或单服务器的子系统间进行通信，是分布式系统标准的配置。

rabbitmq

三、rabbitmq六种工作模式

Ⅰ、简单模式

简单

  • 发送消息的程序是生产者
  • 队列就代表一个邮箱。虽然消息会流经RbbitMQ和你的应用程序，但消息只能被存储在队列里。队列存储空间只受服务器内存和磁盘限制，它本质上是一个大的消息缓冲区。多个生产者可以向同一个队列发送消息，多个消费者也可以从同一个队列接收消息.
  • 消费者等待从队列接收消息

 简单模式

使用过程

1、创建springboot工程，添加依赖：

  1.  
  2. <dependency> 
  3.     <groupId>com.rabbitmq</groupId> 
  4.     <artifactId>amqp-client</artifactId> 
  5.     <version>5.4.3</version> 
  6. </dependency>

2、生产者发送消息

  1.  
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  5. import com.rabbitmq.client.MessageProperties;
  6.  
  7. import java.io.IOException;
  8. import java.util.concurrent.TimeoutException;
  9.  
  10. public class Producer {
  11.  
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  13.         //1、连接
  14.         ConnectionFactory factory = new ConnectionFactory();
  15.         factory.setHost("192.168.64.140");
  16.         factory.setPort(5672);
  17.         factory.setPassword("admin");
  18.         factory.setUsername("admin");
  19.         Connection connection = factory.newConnection();
  20.         Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  21.  
  22.         //2、在服务器上创建helloworld队列（mq里面创建队列）
  23.         /* 注意事项：队列如果已经存在，不会重复创建，对于队列的生产者和消费者，都要创建队列，保证谁先启动谁先创建
  24.         channel.queueDeclare(队列名,
  25.                              是否是持久队列（一直保存在本地）,
  26.                              是否是独占（排他）队列（多个消费者能不能从同一个队列接收消息）false表示不是,
  27.                              是否是自动删除的队列（没有消费者时服务器是否自动删除）false表示不自动删除,
  28.                              队列的其他参数属性，使用键值对方式
  29.                              );*/
  30.         channel.queueDeclare("helloworld",false,false,false,null);
  31.  
  32.  
  33.  
  34.         //3、向helloworld队列发送消息，参数4为发送的消息，并且只能为byte数组
  35.         /*channel.basicPublish(交换机（为空表示使用默认交换机）,
  36.                                队列名字,
  37.                                消息的其他属性配置（键值对设置消息属性，没有使用null值）,
  38.                                队列的内容（byte数组）);
  39.         * */
  40.         channel.basicPublish("","helloworld", MessageProperties.PERSISTENT_BASIC,"hello world".getBytes());
  41.                                                                     //现成的键值对
  42.  
  43.     }
  44. }

注意事项：

消费者与生产者都建立连接的原因，由于相同队列不会重复创建，所以，当谁先启动谁就先创建消息队列，节省时间

3、消费者接收消息

  1. import com.rabbitmq.client.*;
  2. import java.io.IOException;
  3. import java.util.concurrent.TimeoutException;
  4.  
  5. public class Consumer {
  6.  
  7.     public static void main(String[] args) throws IOException, TimeoutException {
  8.         //1、连接
  9.         ConnectionFactory factory = new ConnectionFactory();
  10.         factory.setHost("192.168.64.140");
  11.         factory.setPort(5672);
  12.         factory.setPassword("admin");
  13.         factory.setUsername("admin");
  14.         Connection connection = factory.newConnection();
  15.         Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  16.  
  17.         //2、在服务器上创建helloworld队列（mq里面创建队列）
  18.         //注意事项：队列如果已经存在，不会重复创建
  19.         channel.queueDeclare("helloworld",false,false,false,null);
  20.         
  21.         //3.1创建处理消息的回调对象
  22.         DeliverCallback deliverCallback =new DeliverCallback() {
  23.  
  24.             public void handle(String s, Delivery delivery) throws IOException {
  25.  
  26.                 byte[] a = delivery.getBody(); //获取队列消息，byte类型
  27.                 String s1 = new String(a); //将byte类型转换成字符串类型
  28.             }
  29.         };
  30.  
  31.         //3.2取消消息处理的回调对象
  32.         CancelCallback cancelCallback = new CancelCallback() {
  33.  
  34.             public void handle(String s) throws IOException {
  35.  
  36.             }
  37.         };
  38.         
  39.         /*3、从helloworld队列接收消息，把消息传递到回调对象处理
  40.         channel.basicConsume(队列的名字,
  41.                               确认方式 ACK（Acknowledgment）true自动确认  false手动确认,
  42.                               处理消息的回调对象,
  43.                               取消消息处理的回调对象);*/
  44.         channel.basicConsume("helloworld",true,deliverCallback,cancelCallback);
  45.     }
  46. }

Ⅱ、工作模式

工作

 工作模式

消费者有多个，当生产者发送数据时，多个消费者接收消息默认采用轮询策略，这就导致了当某个消费者处理消息的回调对象内的业务处理不完，造成阻塞，其他闲置的消费者也没办法接收到生产者的信息，所以要更改配置，变为每个消费者处理完自己的消息的回调对象内数据，再接收生产者的信息，避免堵塞。

需要修改如下配置参数：

        ①、设置qos=1 每次3只接收一条信息，处理完之前不接收下一条          

        channel.basicQos(1);//3、设置Qos   每次只收一条，处理完之前不好收下一条

        ②、当处理完消息，告诉服务器自己已经处理完消息  autoAck=false

        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); //2、回执，告诉服务器已经处理完消息

         ③、还要将ACK修改为false手动模式

        其他配置         

        ④、由于数据都在rabbitmq消息中间件，如果该服务器崩溃，会造成传递信息丢失，持

                久化到本地磁盘的方法，该参数为true      

        channel.queueDeclare("helloworld1",true,false,false,null);

注释：这些修改项已经在代码中标注

使用步骤

1、添加如上依赖

  1.  
  2. <dependency> 
  3.     <groupId>com.rabbitmq</groupId> 
  4.     <artifactId>amqp-client</artifactId> 
  5.     <version>5.4.3</version> 
  6. </dependency>

2、生产者发送消息

  1. import com.rabbitmq.client.Channel;
  2. import com.rabbitmq.client.Connection;
  3. import com.rabbitmq.client.ConnectionFactory;
  4. import java.io.IOException;
  5. import java.util.Scanner;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. public class Producer2 {
  9.  
  10.     public static void main(String[] args) throws IOException, TimeoutException {
  11.  
  12.         //1、连接
  13.         ConnectionFactory factory = new ConnectionFactory();
  14.         factory.setHost("192.168.64.140");
  15.         factory.setPort(5672);
  16.         factory.setPassword("admin");
  17.         factory.setUsername("admin");
  18.         Connection connection = factory.newConnection();
  19.         Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  20.  
  21.        /*2、 在服务器上创建helloworld队列（mq里面创建队列）
  22.         注意事项：队列如果已经存在，不会重复创建，对于队列的生产者和消费者，都要创建队列，保证谁先启动谁先创建
  23.         channel.queueDeclare(队列名,
  24.                              是否是持久队列（一直保存在本地）,
  25.                              是否是独占（排他）队列（多个消费者能不能从同一个队列接收消息）false表示不是,
  26.                              是否是自动删除的队列（没有消费者时服务器是否自动删除）false表示不自动删除,
  27.                              队列的其他参数属性，使用键值对方式
  28.                              );*/
  29.         channel.queueDeclare("helloworld1",true,false,false,null);
  30.                                                 //④ ④ ④ ④ ④
  31.         //循环向队列写入信息
  32.         while(true){
  33.             System.out.println("输入消息");
  34.             String s = new Scanner(System.in).nextLine();
  35.  
  36.         //3、向helloworld队列发送消息，参数4为发送的消息，并且只能为byte数组
  37.         /*channel.basicPublish(交换机（为空表示使用默认交换机）,
  38.                                队列名字,
  39.                                消息的其他属性配置（键值对设置消息属性，没有使用null值）,
  40.                                队列的内容（byte数组）);
  41.         * */
  42.             channel.basicPublish("","helloworld1",null,s.getBytes());
  43.         }
  44.     }
  45. }

3、消费者接收消息

  1. import com.rabbitmq.client.*;
  2. import java.io.IOException;
  3. import java.util.concurrent.TimeoutException;
  4.  
  5. public class Consumer2 {
  6.     public static void main(String[] args) throws IOException, TimeoutException {
  7.         //1、连接
  8.         ConnectionFactory factory = new ConnectionFactory();
  9.         factory.setHost("192.168.64.140");
  10.         factory.setPort(5672);
  11.         factory.setPassword("admin");
  12.         factory.setUsername("admin");
  13.         Connection connection = factory.newConnection();
  14.         final Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  15.  
  16.         //2、在服务器上创建helloworld队列（mq里面创建队列）
  17.         //注意事项：队列如果已经存在，不会重复创建
  18.         channel.queueDeclare("helloworld1", true, false, false, null);
  19.  
  20.  
  21.         //3.1、创建处理消息的回调对象
  22.         DeliverCallback deliverCallback = new DeliverCallback() {
  23.  
  24.             public void handle(String s, Delivery delivery) throws IOException {
  25.  
  26.                 String str = new String(delivery.getBody());
  27.                 System.out.println(str);  //打印消息队列
  28.  
  29.                 for (int i = 0; i < str.length(); i++) {
  30.                     if (str.charAt(i) == '.') {
  31.                         try {
  32.                             Thread.sleep(5000);
  33.                         } catch (InterruptedException e) {
  34.                             System.out.println(e);
  35.                         }
  36.                     }
  37.                 }
  38.                 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); //2、回执，处理完成回执数据
  39.                                                                         //② ② ② ② ② ②
  40.             }
  41.         };
  42.  
  43.  
  44.         //3.1、取消消息处理的回调对象
  45.         CancelCallback cancelCallback = new CancelCallback() {
  46.             public void handle(String s) throws IOException {
  47.             }
  48.         };
  49.  
  50.  
  51.         channel.basicQos(1);//3、设置Qos   每次只收一条，处理完之前不好收下一条
  52.         //①①①①①①①①①①①①①①①
  53.  
  54.  
  55.  
  56.         /*3、从helloworld队列接收消息，把消息传递到回调对象处理
  57.         channel.basicConsume(队列的名字,
  58.                               确认方式 ACK（Acknowledgment）true自动确认（如果消息在确认之前消费者崩溃，这条消息作废）  false手动确认（如果消息在确认之前消费者崩溃，客户端消息回滚，重新接收）,
  59.                               处理消息的回调对象,
  60.                               取消消息处理的回调对象);*/
  61.         channel.basicConsume("helloworld1", false, deliverCallback, cancelCallback);
  62.                                                 //③③③③③③③③
  63.     }
  64. }

Ⅲ、发布订阅模式

发布订阅

发布订阅

类似于电台和听众之间的关系 

向多个消费者传递同一条消息。这种模式称为“发布/订阅”。 

为了说明该模式，我们将构建一个简单的日志系统。它将由两个程序组成——第一个程序将发出日志消息，第二个程序接收它们（开启多个实例）。

首次使用到了交换机（Exchanges）

使用步骤

1、生产者发送消息

  1. import com.rabbitmq.client.BuiltinExchangeType;
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  5.  
  6. import java.io.IOException;
  7. import java.util.Scanner;
  8. import java.util.concurrent.TimeoutException;
  9.  
  10. public class Producer3 {
  11.  
  12.     //发布订阅模式（将消息广播出去，所有消费者都能接收到）（如果没有消费者接收，发出去的消息直接失效丢失）
  13.     //模拟各个服务向消费者发送发送，中间mq转发
  14.     public static void main(String[] args) throws IOException, TimeoutException {
  15.  
  16.         //1、连接
  17.         ConnectionFactory factory = new ConnectionFactory();
  18.         factory.setHost("192.168.64.140");
  19.         factory.setPort(5672);
  20.         factory.setPassword("admin");
  21.         factory.setUsername("admin");
  22.         Connection connection = factory.newConnection();
  23.         Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  24.  
  25.  
  26.         //2、创建一个 Fanout 类型的交换机，并且起名字叫logs(交换机在消费者和生产者都要创建)
  27.         //channel.exchangeDeclare(交换机的名字, 交换机的类型);
  28.         channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
  29.  
  30.         //3、循环向队列写入信息
  31.         while(true){
  32.             System.out.println("输入消息");
  33.             String s = new Scanner(System.in).nextLine();
  34.  
  35.             //4、向helloworld队列发送消息，参数4为发送的消息，并且只能为byte数组
  36.             /*channel.basicPublish(交换机（为空表示使用默认交换机）,
  37.                                    队列名字,
  38.                                    消息的其他属性配置（键值对设置消息属性，没有使用null值）,
  39.                                    队列的内容（byte数组）);
  40.             * */
  41.             channel.basicPublish("logs","",null,s.getBytes());
  42.                                               //对于Fanout，参数二写不写都无效
  43.         }
  44.     }
  45. }

2、消费者接收消息

  1. import com.rabbitmq.client.*;
  2. import java.io.IOException;
  3. import java.util.UUID;
  4. import java.util.concurrent.TimeoutException;
  5.  
  6. public class Consumer3 {
  7.     public static void main(String[] args) throws IOException, TimeoutException {
  8.         //1、连接
  9.         ConnectionFactory factory = new ConnectionFactory();
  10.         factory.setHost("192.168.64.140");
  11.         factory.setPort(5672);
  12.         factory.setPassword("admin");
  13.         factory.setUsername("admin");
  14.         Connection connection = factory.newConnection();
  15.         final Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  16.  
  17.         //2、在服务器上创建XXXX随机队列{独占（排他）,自动删除，非持久}
  18.         //注意事项：消费者需要绑定生产者的频段，才能接收到消息。队列的创建采用随机名字（实际工作中消费者成千上百）
  19.         String str = UUID.randomUUID().toString();
  20.         channel.queueDeclare(str, false, true, true, null);
  21.         //3、创建一个 Fanout 类型的交换机，并且起名字叫logs
  22.         channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
  23.         //4、交换机和队列绑定
  24.         channel.queueBind(str,"logs","");
  25.  
  26.  
  27.  
  28.         //5.1、创建处理消息的回调对象
  29.         DeliverCallback deliverCallback = new DeliverCallback() {
  30.  
  31.             public void handle(String s, Delivery delivery) throws IOException {
  32.                 String str = new String(delivery.getBody());  //接收队列里的消息
  33.                 System.out.println("收到" + str);  //处理消息
  34.             }
  35.         };
  36.         //5.2取消消息处理的回调对象
  37.         CancelCallback cancelCallback = new CancelCallback() {
  38.             public void handle(String s) throws IOException {
  39.             }
  40.         };
  41.  
  42.  
  43.         /*5、从随机队列接收消息，把消息传递到回调对象处理
  44.         channel.basicConsume(队列的名字,
  45.                               确认方式 ACK（Acknowledgment）true自动确认（如果消息在确认之前消费者崩溃，这条消息作废）  false手动确认（如果消息在确认之前消费者崩溃，客户端消息回滚，重新接收）,
  46.                               处理消息的回调对象,
  47.                               取消消息处理的回调对象);*/
  48.         channel.basicConsume(str, true, deliverCallback, cancelCallback);
  49.  
  50.     }
  51. }

Ⅳ、路由模式

生产者携带路由关键字与接收的消费者关键字做比对，消费者决定该消息是否接收

路由

路由模式

使用步骤

1、生产者发送消息

  1. import com.rabbitmq.client.BuiltinExchangeType;
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  5.  
  6. import java.io.IOException;
  7. import java.sql.SQLOutput;
  8. import java.util.Scanner;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. public class Producer4 {
  12.  
  13.     //路由模式根据绑定的关键词匹配
  14.     //模拟各个服务向消费者发送发送日志，携带关键字，中间mq转发
  15.     public static void main(String[] args) throws IOException, TimeoutException {
  16.  
  17.         //1、连接
  18.         ConnectionFactory factory = new ConnectionFactory();
  19.         factory.setHost("192.168.64.140");
  20.         factory.setPort(5672);
  21.         factory.setPassword("admin");
  22.         factory.setUsername("admin");
  23.         factory.setVirtualHost("kongjuian1");  //指定使用空间名字，默认空间名为 “/”
  24.         Connection connection = factory.newConnection();
  25.         Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  26.  
  27.  
  28.         //2、创建一个 Direct 类型的交换机，并且起名字叫direct_logs(交换机在消费者和生产者都要创建)
  29.         channel.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT);
  30.  
  31.         //3、发送消息，消息上携带关键字信息
  32.         while(true){
  33.             System.out.println("输入消息");
  34.             String s = new Scanner(System.in).nextLine();
  35.             System.out.println("请输入路由键： ");
  36.             String k = new  Scanner(System.in).nextLine();
  37.             //4、向helloworld队列发送消息，参数4为发送的消息，并且只能为byte数组
  38.             /*channel.basicPublish(交换机（为空表示使用默认交换机）,
  39.                                     队列名字,现在表示为携带的关键词
  40.                                     消息的其他属性配置（键值对设置消息属性，没有使用null值）,
  41.                                     队列的内容（byte数组）);
  42.             * */
  43.             channel.basicPublish("direct_logs",k,null,s.getBytes());
  44.         }
  45.     }
  46. }

2、消费者接收消息

  1. import com.rabbitmq.client.*;
  2.  
  3. import java.io.IOException;
  4. import java.util.Scanner;
  5. import java.util.UUID;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. public class Consumer4 {
  9.     public static void main(String[] args) throws IOException, TimeoutException {
  10.         //1、连接
  11.         ConnectionFactory factory = new ConnectionFactory();
  12.         factory.setHost("192.168.64.140");
  13.         factory.setPort(5672);
  14.         factory.setPassword("admin");
  15.         factory.setUsername("admin");
  16.         factory.setVirtualHost("kongjuian1");  //指定使用空间名字，默认空间名为 “/”
  17.         Connection connection = factory.newConnection();
  18.         final Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  19.  
  20.         //2、在服务器上创建XXXX随机队列{独占（排他）,自动删除，非持久}
  21.         //让服务器自动命名，自动提供队列参数
  22.         //channel.queueDeclare(随机队列名, false, true, true, null);
  23.         channel.queueDeclare(); //服务器自动提供的参数为上面所示参数
  24.         String name = channel.queueDeclare().getQueue();  //得到服务器创建的随机名字
  25.         //3、创建一个 DIRECT 类型的交换机，并且起名字叫logs
  26.         channel.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT);
  27.  
  28.         //4、设置绑定键
  29.         System.out.println("请输入绑定键，用空格隔开： ");//aa bb cc  获取到，根据空格拆分放到一个数组里面
  30.         String s = new  Scanner(System.in).nextLine();
  31.         String[] a =  s.split("\\s+"); //    \s是空白字符，相当于空格  +表示1到多个
  32.  
  33.         for (String k : a){
  34.             channel.queueBind(name,"direct_logs",k);
  35.         }
  36.  
  37.  
  38.  
  39.         //5.1、创建处理消息的回调对象
  40.         DeliverCallback deliverCallback = new DeliverCallback() {
  41.  
  42.             public void handle(String s, Delivery delivery) throws IOException {
  43.                 String str = new String(delivery.getBody());  //接收队列里的消息
  44.                 System.out.println("收到" + str);  //处理消息
  45.             }
  46.         };
  47.         //5.2取消消息处理的回调对象
  48.         CancelCallback cancelCallback = new CancelCallback() {
  49.             public void handle(String s) throws IOException {
  50.             }
  51.         };
  52.  
  53.  
  54.         /*5、从随机队列接收消息，把消息传递到回调对象处理
  55.         channel.basicConsume(队列的名字,
  56.                               确认方式 ACK（Acknowledgment）true自动确认（如果消息在确认之前消费者崩溃，这条消息作废）  false手动确认（如果消息在确认之前消费者崩溃，客户端消息回滚，重新接收）,
  57.                               处理消息的回调对象,
  58.                               取消消息处理的回调对象);*/
  59.         channel.basicConsume(name, true, deliverCallback, cancelCallback);
  60.  
  61.     }
  62. }

mq里面可以创建不同的空间，不同空间相互隔离，互不影响

创建方法：在服务器上创建

  factory.setVirtualHost("kongjuian1");  //指定使用空间名字，默认空间名为 “/” 

Ⅴ、主题模式

主题

主题模式可以理解为路由模式的升级，绑定的路由关键字，可以使用通配符代替（类似于正则表达式）


*通配一个字符            #通配多个字符
 

使用步骤

1、生产者发送消息

  1. import com.rabbitmq.client.BuiltinExchangeType;
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  5.  
  6. import java.io.IOException;
  7. import java.util.Scanner;
  8. import java.util.concurrent.TimeoutException;
  9.  
  10. public class Producer5 {
  11.  
  12.     //主题模式根据绑定的关键词匹配（和路由的区别在于关键词格式，和交换机不一样）
  13.     public static void main(String[] args) throws IOException, TimeoutException {
  14.  
  15.         //1、连接
  16.         ConnectionFactory factory = new ConnectionFactory();
  17.         factory.setHost("192.168.64.140");
  18.         factory.setPort(5672);
  19.         factory.setPassword("admin");
  20.         factory.setUsername("admin");
  21.         factory.setVirtualHost("kongjuian1");  //指定使用空间名字，默认空间名为 “/”
  22.         Connection connection = factory.newConnection();
  23.         Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  24.  
  25.  
  26.         //2、创建一个 Topic 类型的交换机，并且起名字叫direct_logs(交换机在消费者和生产者都要创建)
  27.         channel.exchangeDeclare("Topic_logs", BuiltinExchangeType.TOPIC);
  28.  
  29.         //3、发送消息，消息上携带关键字信息
  30.         while(true){
  31.             System.out.println("输入消息");
  32.             String s = new Scanner(System.in).nextLine();
  33.             System.out.println("请输入路由键： ");
  34.             String k = new  Scanner(System.in).nextLine();
  35.             //4、向helloworld队列发送消息，参数4为发送的消息，并且只能为byte数组
  36.             /*channel.basicPublish(交换机（为空表示使用默认交换机）,
  37.                                     队列名字,现在表示为携带的关键词
  38.                                     消息的其他属性配置（键值对设置消息属性，没有使用null值）,
  39.                                     队列的内容（byte数组）);
  40.             * */
  41.             channel.basicPublish("Topic_logs",k,null,s.getBytes());
  42.         }
  43.     }
  44. }

2、消费者接收消息

  1. import com.rabbitmq.client.*;
  2.  
  3. import java.io.IOException;
  4. import java.util.Scanner;
  5. import java.util.concurrent.TimeoutException;
  6.  
  7. public class Consumer5 {
  8.     public static void main(String[] args) throws IOException, TimeoutException {
  9.         //1、连接
  10.         ConnectionFactory factory = new ConnectionFactory();
  11.         factory.setHost("192.168.64.140");
  12.         factory.setPort(5672);
  13.         factory.setPassword("admin");
  14.         factory.setUsername("admin");
  15.         factory.setVirtualHost("kongjuian1");  //指定使用空间名字，默认空间名为 “/”
  16.         Connection connection = factory.newConnection();
  17.         final Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
  18.  
  19.         //2、在服务器上创建XXXX随机队列{独占（排他）,自动删除，非持久}
  20.         //让服务器自动命名，自动提供队列参数
  21.         //channel.queueDeclare(随机队列名, false, true, true, null);
  22.         channel.queueDeclare(); //服务器自动提供的参数为上面所示参数
  23.         String name = channel.queueDeclare().getQueue();  //得到服务器创建的随机名字
  24.         //3、创建一个 DIRECT 类型的交换机，并且起名字叫logs
  25.         channel.exchangeDeclare("Topic_logs", BuiltinExchangeType.TOPIC);
  26.  
  27.         //4、设置绑定键
  28.         System.out.println("请输入绑定键，用空格隔开： ");//aa.ab.bc bb.bb cc.dd.gg  获取到，根据空格拆分放到一个数组里面
  29.         String s = new  Scanner(System.in).nextLine();
  30.         String[] a =  s.split("\\s+"); //    \s是空白字符，相当于空格  +表示1到多个
  31.  
  32.         for (String k : a){
  33.             channel.queueBind(name,"Topic_logs",k);
  34.         }
  35.  
  36.  
  37.  
  38.         //5.1、创建处理消息的回调对象
  39.         DeliverCallback deliverCallback = new DeliverCallback() {
  40.  
  41.             public void handle(String s, Delivery delivery) throws IOException {
  42.                 String str = new String(delivery.getBody());  //接收队列里的消息
  43.                 System.out.println("收到" + str);  //处理消息
  44.             }
  45.         };
  46.         //5.2取消消息处理的回调对象
  47.         CancelCallback cancelCallback = new CancelCallback() {
  48.             public void handle(String s) throws IOException {
  49.             }
  50.         };
  51.  
  52.  
  53.         /*5、从随机队列接收消息，把消息传递到回调对象处理
  54.         channel.basicConsume(队列的名字,
  55.                               确认方式 ACK（Acknowledgment）true自动确认（如果消息在确认之前消费者崩溃，这条消息作废）  false手动确认（如果消息在确认之前消费者崩溃，客户端消息回滚，重新接收）,
  56.                               处理消息的回调对象,
  57.                               取消消息处理的回调对象);*/
  58.         channel.basicConsume(name, true, deliverCallback, cancelCallback);
  59.  
  60.     }
  61. }

四、消息服务案例


 

前两个案例的地址：(1条消息) Spring cloud Netflix_asddasddeedd的博客-CSDN博客

  1. BUS配置刷新 刷新指令消息发送到Rabbitmq ，其他模块接收执行，执行刷新操作 主题模式
  2. sleuth + zipkin 链路跟踪 链路跟踪日志，通过 Rabbitmq 中转发送到 Zipkin 服务器 简单模式
  3. 订单的流量削峰 购物系统产生的订单，不直接存储到数据库，而是发送到 Rabbitmq， 后台的消费者模块接收订单，再向数据库存储， 短时间大量订单并发存储，变成顺序存储 简单模式 / 工作模式。

案例三：

案例简介：

订单的流量削峰 购物系统产生的订单，不直接存储到数据库，而是发送到 Rabbitmq， 后台的消费者模块接收订单，再向数据库存储， 短时间大量订单并发存储，变成顺序存储 简单模式 / 工作模式。


订单发送到 Rabbitmq

1、添加 Rabbitmq 依赖和 Rabbitmq 连接配置

  1. <dependency>
  2. 	<groupId>org.springframework.boot</groupId>
  3. 	<artifactId>spring-boot-starter-amqp</artifactId>
  4. </dependency>
  1. spring：
  2.   rabbitmq:
  3.     host: 192.168.64.140
  4.     port: 5672
  5.     username: admin
  6.     password: admin
  7.     virtual-host: /

2、在启动类（或者自定义自动配置类）中设置队列的参数： orderQueue, true, false, false

  1. import org.mybatis.spring.annotation.MapperScan;
  2. import org.springframework.amqp.core.Queue;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5. import org.springframework.context.annotation.Bean;
  6.  
  7.  
  8. @SpringBootApplication
  9. @MapperScan("com.pd.mapper")
  10. public class RunPdAPP{
  11. 	
  12. 	public static void main(String[] args) {
  13. 		SpringApplication.run(RunPdAPP.class, args);
  14. 	}
  15.  
  16.  
  17. 	//新建spring的Queue实例用来封装队列的参数
  18. 	//rabbitmq的自动配置类会自动发现这个Queue实例
  19. 	//根据其中的参数自动在服务器上创建队列
  20. 	@Bean
  21. 	public Queue orderQueue(){//持久，非独占，不自动删除
  22. 		return  new Queue("orderQueue",true,false,false);
  23. 	}
  24.  
  25. }

 3、修改需要发送消息的模块

  1. 	@Autowired
  2. 	private AmqpTemplate t;  //注入对象： AmqpTemplate（用来封装发送消息代码的工具）发送消息
  3.  
  4.  
  5. 	public String saveOrder(PdOrder pdOrder) throws Exception {
  6. 		String orderId = generateId(); // 生成订单id（该方法随机生成时间作为id订单编号）
  7. 		pdOrder.setOrderId(orderId); // 订单id放入订单对象
  8. 		// 订单id、地址id、用户id、购买的商品id
  9.  
  10. 		// 转换并发送，先把数据转成byte[]再发送
  11. 		t.convertAndSend("orderQueue", pdOrder);


订单的消费者

1、Rabbitmq基础配置，准备队列参数

        ①、添加 Rabbitmq 依赖和 Rabbitmq 连接配置

        ③、在启动类（或者自定义自动配置类）中设置队列的参数： orderQueue, true, false, false


2、新建消费者类： OrderConsumer
3、用注解配置接收消息

  1. import com.pd.pojo.PdOrder;
  2. import com.pd.service.OrderService;
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.stereotype.Component;
  7.                         //队列名字
  8. @RabbitListener(queues = "orderQueue") //通过注解配置就可以接收信息，不需要写代码
  9. @Component
  10. public class OrderConsumer {
  11.  
  12.     @Autowired
  13.     private OrderService orderService;
  14.  
  15.     @RabbitHandler //指定使用哪个方法处理该队列消息，并且该方法只能有一个
  16.     public void  receive(PdOrder order) throws Exception {
  17.         orderService.saveOrder(order);
  18.         System.out.println("=============================订单已经完成==================");
  19.     }
  20.  
  21. }


4、将接收到的消息储存到数据库

五、RabbitMQ - Spring boot 整合

 添加RabbitMQ依赖

  1.         <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-amqp</artifactId>
  4.         </dependency>

配置yml，配置RabbitMQ连接信息

  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.64.140
  4.     port: 5672
  5.     username: admin
  6.     password: admin
  7.     virtual-host: /

1、简单模式整合

①、配置启动类

  1. import org.springframework.amqp.core.Queue;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5. import org.springframework.context.annotation.Bean;
  6.  
  7. import javax.annotation.PostConstruct;
  8.  
  9. @SpringBootApplication
  10. public class Main1 {
  11.  
  12.     public static void main(String[] args) {
  13.         SpringApplication.run(Main1.class,args);
  14.     }
  15.  
  16.  
  17.     //设置队列参数
  18.     @Bean
  19.     public Queue helloWorld(){
  20.                                 //参数简介 ：队列名称，非持久，非独占，不自动删除
  21.         return new Queue("helloworld",false,false,false);
  22.     }
  23.  
  24.  
  25.     //调用生产者发送消息
  26.     @Autowired
  27.     private Producer1 producer;
  28.  
  29.     //spring的执行流程
  30.     //自动扫描创建实例--》完成依赖注入--》@PostConstruct-->后续步骤
  31.     @PostConstruct
  32.     public void test(){
  33.  
  34.         //开启一个新的线程，不阻塞程序主线程执行
  35.         new Thread(new Runnable() {
  36.             @Override
  37.             public void run() {
  38.                 try {
  39.                     Thread.sleep(3000);  //等待helloworld队列被创建
  40.                 } catch (InterruptedException e) {
  41.                     e.printStackTrace();
  42.                 }
  43.                 producer.send();
  44.             }
  45.         }).start();
  46.     }
  47. }

②、创建生产者类

  1. import org.springframework.amqp.core.AmqpTemplate;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.stereotype.Component;
  4.  
  5. @Component
  6. public class Producer1 {
  7.  
  8.     //发送消息的封装工具
  9.     //
  10.     @Autowired
  11.     private AmqpTemplate t;
  12.     
  13.     public void send(){
  14.         //向helloworld队列发送消息          //自动转换成byter数组
  15.         //相当于： channel.basicPublish("logs","",null,s.getBytes());
  16.         t.convertAndSend("helloworld","hello world");
  17.     }
  18. }

③、创建消费者类

  1. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  2. import org.springframework.stereotype.Component;
  3.  
  4. //@RabbitListener(queues = "helloworld")
  5. //接收队列helloworld,
  6. //如果放在类上，要与 @RabbitHandler配合使用
  7. //也可以直接梵高方法上。
  8.  
  9. @Component
  10. public class Consumer1 {
  11.  
  12.     @RabbitListener(queues = "helloworld")
  13.     public void receive(String msg){
  14.         System.out.println("收到消息:" + msg);
  15.     }
  16.  
  17. }

2、工作模式

修改yml，每次抓取消息修改为1

  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.64.140
  4.     port: 5672
  5.     username: admin
  6.     password: admin
  7.     virtual-host: /
  8.     listener:
  9.       simple:
  10.         prefetch: 1    #每次抓取一条消息，处理完之前不收下一条，默认抓取250条

①、配置启动类

  1. import org.springframework.amqp.core.Queue;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5. import org.springframework.context.annotation.Bean;
  6.  
  7. import javax.annotation.PostConstruct;
  8. import java.util.Scanner;
  9.  
  10. /*
  11.     合理分发
  12.           1. autoAck=false，手动确认
  13.              spring封装的rabbitmq，默认就是手动确认，
  14.              spring会自动执行发送回执
  16.           2. qos=1，预抓取消息数量（配置类中修改）
  17.              yml 添加 pre-fetch=1，默认是250
  19.     消息持久化
  20.           1. 队列持久化
  21.              new Queue("", true)
  23.           2. 消息数据的持久化
  24.              spring发送的消息，默认就是持久消息
  25.  */
  26.  
  27. @SpringBootApplication
  28. public class Main2 {
  29.  
  30.     public static void main(String[] args) {
  31.         SpringApplication.run(Main2.class, args);
  32.     }
  33.  
  34.     // 设置队列参数
  35.     @Bean
  36.     public Queue taskQueue() {
  37.         // 持久，非独占，不自动删除
  38.         return new Queue("task_queue"); //只给队列名，其他参数是下面的默认值
  39.         // return new Queue("task_queue",true,false,false);
  40.     }
  41.  
  42.     @Autowired
  43.     private Producer2 p;
  44.     /*
  45.     spring的主线程执行流程
  46.     自动扫描创建实例 --> 完成依赖注入 --> @PostConstruct --> 后续步骤
  47.      */
  48.     @PostConstruct
  49.     public void test() {
  50.         // 在新线程中执行自己的运算，不阻塞 spring 主线程执行
  51.         new Thread(() -> {
  52.             while (true){
  53.                 System.out.print("输入消息：");
  54.                 String s = new Scanner(System.in).nextLine();
  55.                 p.send(s);
  56.             }
  57.         }).start();
  58.     }
  59.  
  60. }

②创建生产者类

  1. import org.springframework.amqp.core.AmqpTemplate;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.stereotype.Component;
  4.  
  5.  
  6. @Component
  7. public class Producer2 {
  8.     // 发送消息的封装工具
  9.     // RabbitAutoConfiguration 中创建了 AmqpTemplate 实例
  10.     @Autowired
  11.     private AmqpTemplate t;
  12.  
  13.     public void send(String s) {
  14.         // 向 helloworld 队列发送消息
  15.         // 队列的参数在启动类中设置
  16.         t.convertAndSend("task_queue", s);
  17.  
  18.         /*
  19.         t.convertAndSend("task_queue", s, 消息预处理对象);
  20.         在预处理对象中，可以对消息的参数进行调整，
  21.         可以把持久化参数设置成非持久
  22.          */
  23.     }
  24. }

③创建消费者类

  1. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  2. import org.springframework.stereotype.Component;
  3.  
  4.  
  5. //每个@RabbitListener都会注册成为一个消费者
  6. @Component
  7. public class Consumer2 {
  8.     @RabbitListener(queues = "task_queue")
  9.     public void receive1(String msg) {
  10.         System.out.println("消费者1收到："+msg);
  11.     }
  12.  
  13.     @RabbitListener(queues = "task_queue")
  14.     public void receive2(String msg) {
  15.         System.out.println("消费者2收到："+msg);
  16.     }
  17. }

3、发布订阅模式

启动类

  1. import org.springframework.amqp.core.FanoutExchange;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5. import org.springframework.context.annotation.Bean;
  6.  
  7. import javax.annotation.PostConstruct;
  8. import java.util.Scanner;
  9.  
  10. @SpringBootApplication
  11. public class Main3 {
  12.  
  13.     public static void main(String[] args) {
  14.         SpringApplication.run(Main3.class,args);
  15.     }
  16.  
  17.  
  18.     //创建FANOUT交换机
  19.     @Bean
  20.     public FanoutExchange logs(){
  21.         return new FanoutExchange("logs",false,false);
  22.     }
  23.  
  24.  
  25.     //开始调用生产者发送消息
  26.     @Autowired
  27.     private Producer3 producer3;
  28.  
  29.     //spring的执行流程
  30.     //自动扫描创建实例--》完成依赖注入--》@PostConstruct-->后续步骤
  31.     @PostConstruct
  32.     public void test(){
  33.  
  34.         //开启一个新的线程，不阻塞程序主线程执行
  35.         new Thread(new Runnable() {
  36.             @Override
  37.             public void run() {
  38.                 while(true){
  39.                     System.out.println("输入消息： ");
  40.                     String s = new Scanner(System.in).nextLine();
  41.                     producer3.send(s);
  42.                 }
  43.             }
  44.         }).start();
  45.     }
  46. }

生产者

  1. import org.springframework.amqp.core.AmqpTemplate;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.stereotype.Component;
  4.  
  5. @Component
  6. public class Producer3 {
  7.  
  8.     //发送消息的封装工具
  9.     @Autowired
  10.     private AmqpTemplate t;
  11.  
  12.  
  13.     public void send(String s){
  14.         //向helloworld队列发送消息     //自动转换成byter数组
  15.         t.convertAndSend("logs","",s);
  16.         //相当于： channel.basicPublish("logs","",null,s.getBytes());
  17.     }
  18. }

消费者

  1. import org.springframework.amqp.rabbit.annotation.Exchange;
  2. import org.springframework.amqp.rabbit.annotation.Queue;
  3. import org.springframework.amqp.rabbit.annotation.QueueBinding;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  6.  
  7. @Component
  8. public class Consumer3 {
  9.  
  10.  
  11.     //创建队列，并且和交换机进行绑定
  12.     //参数为绑定配置
  13.     @RabbitListener(bindings = @QueueBinding(
  14.             value = @Queue,   //创建队列,该注解不给参数表示创建随机队列，
  15.             exchange = @Exchange(name = "logs" ,declare = "false")  //设置交换机
  16.                                                     //false表示不创建新的交换机，使用已经存在的交换机
  17.     ))
  18.     public void receive1(String msg){
  19.         System.out.println("1号收到消息:" + msg);
  20.     }
  21.  
  22.     
  23.     @RabbitListener(bindings = @QueueBinding(
  24.             value = @Queue,
  25.             exchange = @Exchange(name = "logs" ,declare = "false")
  26.     ))
  27.     public void receive2(String msg){
  28.         System.out.println("2号收到消息:" + msg);
  29.     }
  30. }

4、主题模式

启动类

  1. import org.springframework.amqp.core.DirectExchange;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5. import org.springframework.context.annotation.Bean;
  6.  
  7. import javax.annotation.PostConstruct;
  8. import java.util.Scanner;
  9.  
  10. @SpringBootApplication
  11. public class Main4 {
  12.  
  13.     public static void main(String[] args) {
  14.         SpringApplication.run(Main4.class,args);
  15.     }
  16.  
  17.  
  18.     //创建Direct交换机
  19.     @Bean
  20.     public DirectExchange logs(){
  21.         return new DirectExchange("direct_log",false,false);
  22.     }
  23.  
  24.  
  25.     //开始调用生产者发送消息
  26.     @Autowired
  27.     private Producer4 producer4;
  28.  
  29.     //spring的执行流程
  30.     //自动扫描创建实例--》完成依赖注入--》@PostConstruct-->后续步骤
  31.     @PostConstruct
  32.     public void test(){
  33.  
  34.         //开启一个新的线程，不阻塞程序主线程执行
  35.         new Thread(new Runnable() {
  36.             @Override
  37.             public void run() {
  38.                 while(true){
  39.                     System.out.println("输入消息： ");
  40.                     String s = new Scanner(System.in).nextLine();
  41.                     System.out.println("输入路由键： ");
  42.                     String k = new Scanner(System.in).nextLine();
  43.                     producer4.send(k,s);
  44.                 }
  45.             }
  46.         }).start();
  47.     }
  48. }

生产者

  1. import org.springframework.amqp.core.AmqpTemplate;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.stereotype.Component;
  4.  
  5. @Component
  6. public class Producer4 {
  7.  
  8.     //发送消息的封装工具
  9.     @Autowired
  10.     private AmqpTemplate t;
  11.  
  12.  
  13.     public void send(String k , String s){
  14.  
  15.         //向队列发送消息     //自动转换成byter数组
  16.         //相当于： channel.basicPublish("logs","",null,s.getBytes());
  17.         t.convertAndSend("direct_log",k,s);
  18.  
  19.     }
  20. }

消费者

  1. import org.springframework.amqp.rabbit.annotation.Exchange;
  2. import org.springframework.amqp.rabbit.annotation.Queue;
  3. import org.springframework.amqp.rabbit.annotation.QueueBinding;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  6.  
  7. @Component
  8. public class Consumer4 {
  9.  
  10.  
  11.     //创建队列，并且和交换机进行绑定
  12.     //参数为绑定配置
  13.     @RabbitListener(bindings = @QueueBinding(
  14.             value = @Queue,   //创建队列,该注解不给参数表示创建随机队列，
  15.             exchange = @Exchange(name = "direct_log" ,declare = "false"),  //设置交换机
  16.                                                     //表示不创建新的交换机，使用已经存在的交换机
  17.             key = {"error"}
  18.     ))
  19.     public void receive1(String msg){
  20.         System.out.println("1号收到消息:" + msg);
  21.     }
  22.  
  23.  
  24.     @RabbitListener(bindings = @QueueBinding(
  25.             value = @Queue,
  26.             exchange = @Exchange(name = "direct_log" ,declare = "false"),
  27.             key = {"info","error","warning"}
  28.     ))
  29.     public void receive2(String msg){
  30.         System.out.println("2号收到消息:" + msg);
  31.     }
  32. }

5、路由模式

启动类

  1. import org.springframework.amqp.core.TopicExchange;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5. import org.springframework.context.annotation.Bean;
  6.  
  7. import javax.annotation.PostConstruct;
  8. import java.util.Scanner;
  9.  
  10. @SpringBootApplication
  11. public class Main5 {
  12.  
  13.     public static void main(String[] args) {
  14.         SpringApplication.run(Main5.class,args);
  15.     }
  16.  
  17.  
  18.     //创建Topic交换机
  19.     @Bean
  20.     public TopicExchange topict(){
  21.         return new TopicExchange("topict_log",false,false);
  22.     }
  23.  
  24.  
  25.     //开始调用生产者发送消息
  26.     @Autowired
  27.     private Producer5 producer5;
  28.  
  29.     //spring的执行流程
  30.     //自动扫描创建实例--》完成依赖注入--》@PostConstruct-->后续步骤
  31.     @PostConstruct
  32.     public void test(){
  33.  
  34.         //开启一个新的线程，不阻塞程序主线程执行
  35.         new Thread(new Runnable() {
  36.             @Override
  37.             public void run() {
  38.                 while(true){
  39.                     System.out.println("输入消息： ");
  40.                     String s = new Scanner(System.in).nextLine();
  41.                     System.out.println("输入路由键： ");
  42.                     String k = new Scanner(System.in).nextLine();
  43.                     producer5.send(k,s);
  44.                 }
  45.             }
  46.         }).start();
  47.     }
  48. }

生产者

  1. import org.springframework.amqp.core.AmqpTemplate;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.stereotype.Component;
  4.  
  5. @Component
  6. public class Producer5 {
  7.  
  8.     //发送消息的封装工具
  9.     @Autowired
  10.     private AmqpTemplate t;
  11.  
  12.  
  13.     public void send(String k , String s){
  14.  
  15.         //向队列发送消息     //自动转换成byter数组
  16.         //相当于： channel.basicPublish("logs","",null,s.getBytes());
  17.         t.convertAndSend("topic_log",k,s);
  18.  
  19.     }
  20. }

消费者

  1. import org.springframework.amqp.rabbit.annotation.Exchange;
  2. import org.springframework.amqp.rabbit.annotation.Queue;
  3. import org.springframework.amqp.rabbit.annotation.QueueBinding;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  6.  
  7. @Component
  8. public class Consumer5 {
  9.  
  10.  
  11.     //创建队列，并且和交换机进行绑定
  12.     //参数为绑定配置
  13.     @RabbitListener(bindings = @QueueBinding(
  14.             value = @Queue,   //创建队列,该注解不给参数表示创建随机队列，
  15.             exchange = @Exchange(name = "topic_log" ,declare = "false"),  //设置交换机
  16.                                                     //表示不创建新的交换机，使用已经存在的交换机
  17.             key = {"*.orange.*"}
  18.     ))
  19.     public void receive1(String msg){
  20.         System.out.println("1号收到消息:" + msg);
  21.     }
  22.  
  23.  
  24.     @RabbitListener(bindings = @QueueBinding(
  25.             value = @Queue,
  26.             exchange = @Exchange(name = "topict_log" ,declare = "false"),
  27.             key = {"*.*.rabbit","lazy.#"}
  28.     ))
  29.     public void receive2(String msg){
  30.         System.out.println("2号收到消息:" + msg);
  31.     }
  32. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/酷酷是懒虫/article/detail/769746
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号