• SpringBoot整合RabbitMQ

    RabbitMQ安装部署详情可见：RabbitMQ简介及在Linux中安装部署（yum）

     一、导入pom.xml依赖

    1. <dependency>
    2.       <groupId>com.rabbitmqgroupId>
    3.       <artifactId>amqp-clientartifactId>
    4.       <version>5.8.0version>
    5. dependency>
    6. <dependency>
    7.       <groupId>org.slf4jgroupId>
    8.       <artifactId>slf4j-nopartifactId>
    9.       <version>1.7.29version>
    10. dependency>

    二、入门测试案例

      发送消息类：Send.java

    1. public class Send {
    2.  
    3.     private final static String QUEUE_NAME = "hello";
    4.  
    5.     public static void main(String[] args) throws IOException, TimeoutException {
    6.         //创建连接工厂
    7.         ConnectionFactory factory = new ConnectionFactory();
    8.         //设置RabbitMQ地址
    9.         factory.setHost("192.168.119.129");
    10.         factory.setUsername("admin");
    11.         factory.setPassword("password");
    12.         //建立连接
    13.         Connection connection = factory.newConnection();
    14.         //获得信道
    15.         Channel channel = connection.createChannel();
    16.         //声明队列
    17.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    18.         //发布消息
    19.         String message = "你好，老6";
    20.         channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
    21.         System.out.println("发送了消息：" + message);
    22.         //关闭连接
    23.         channel.close();
    24.         connection.close();
    25.     }
    26. }

      接收消息类：Recv.java

    1. public class Recv {
    2.     private final static String QUEUE_NAME = "hello";
    3.  
    4.     public static void main(String[] args) throws IOException, TimeoutException {
    5.         //创建连接工厂
    6.         ConnectionFactory factory = new ConnectionFactory();
    7.         //设置RabbitMQ地址
    8.         factory.setHost("192.168.119.129");
    9.         factory.setUsername("admin");
    10.         factory.setPassword("password");
    11.         //建立连接
    12.         Connection connection = factory.newConnection();
    13.         //获得信道
    14.         Channel channel = connection.createChannel();
    15.         //声明队列
    16.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    17.         //接收消息并消费
    18.         channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel){
    19.             @Override
    20.             public void handleDelivery(String consumerTag, Envelope envelope,
    21.                     BasicProperties properties, byte[] body) throws IOException {
    22.                 String message = new String(body, "UTF-8");
    23.                 System.out.println("收到消息：" + message);
    24.             }
    25.         });
    26.     }
    27. }

    运行结果：            ​​​​​

     三、多任务消息队列案例

    发送多任务消息类NewTask.java

    1. public class NewTask {
    2.  
    3.     private final static String TASK_QUEUE_NAME = "task_queue";
    4.  
    5.     public static void main(String[] args) throws IOException, TimeoutException {
    6.         //创建连接工厂
    7.         ConnectionFactory factory = new ConnectionFactory();
    8.         //设置RabbitMQ地址
    9.         factory.setHost("192.168.119.129");
    10.         factory.setUsername("admin");
    11.         factory.setPassword("password");
    12.         //建立连接
    13.         Connection connection = factory.newConnection();
    14.         //获得信道
    15.         Channel channel = connection.createChannel();
    16.         //声明队列
    17.         channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
    18.         for (int i = 0; i < 10; i++) {
    19.             String message;
    20.             if (i % 2 == 0) {
    21.                 message = i + "...";
    22.             } else {
    23.                 message = String.valueOf(i);
    24.             }
    25.             channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));
    26.             System.out.println("发送了消息：" + message);
    27.         }
    28.  
    29.         channel.close();
    30.         connection.close();
    31.     }
    32. }

     接收多任务消息类Worker.java

    1. public class Worker {
    2.  
    3.     private final static String TASK_QUEUE_NAME = "task_queue";
    4.  
    5.     public static void main(String[] args) throws IOException, TimeoutException {
    6.         //创建连接工厂
    7.         ConnectionFactory factory = new ConnectionFactory();
    8.         //设置RabbitMQ地址
    9.         factory.setHost("192.168.119.129");
    10.         factory.setUsername("admin");
    11.         factory.setPassword("password");
    12.         //建立连接
    13.         Connection connection = factory.newConnection();
    14.         //获得信道
    15.         final Channel channel = connection.createChannel();
    16.         //声明队列
    17.         channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
    18.         System.out.println("开始接收消息");
    19.         channel.basicQos(1);
    20.  
    21.         channel.basicConsume(TASK_QUEUE_NAME, false, new DefaultConsumer(channel) {
    22.             @Override
    23.             public void handleDelivery(String consumerTag, Envelope envelope,
    24.                     BasicProperties properties, byte[] body) throws IOException {
    25.                 String message = new String(body, "UTF-8");
    26.                 System.out.println("收到了消息：" + message);
    27.                 try {
    28.                     doWork(message);
    29.                 } finally {
    30.                     System.out.println("消息处理完成");
    31.                     channel.basicAck(envelope.getDeliveryTag(), false);
    32.                 }
    33.             }
    34.         });
    35.     }
    36.  
    37.     private static void doWork(String task) {
    38.         char[] chars = task.toCharArray();
    39.         for (char ch : chars) {
    40.             if (ch == '.') {
    41.                 try {
    42.                     Thread.sleep(1000);
    43.                 } catch (InterruptedException e) {
    44.                     e.printStackTrace();
    45.                 }
    46.             }
    47.         }
    48.     }
    49. }

    运行结果：

    四、接收指定消息案例

    发送消息类：EmitLogDirect.java

    1. public class EmitLogDirect {
    2.  
    3.     private static final String EXCHANGE_NAME = "direct_logs";
    4.  
    5.     public static void main(String[] args) throws IOException, TimeoutException {
    6.         ConnectionFactory factory = new ConnectionFactory();
    7.         factory.setHost("192.168.119.129");
    8.         factory.setUsername("admin");
    9.         factory.setPassword("password");
    10.         Connection connection = factory.newConnection();
    11.         Channel channel = connection.createChannel();
    12.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    13.  
    14.         String message = "info：Hello World!";
    15.         channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes("UTF-8"));
    16.         System.out.println("发送了消息，" + "等级为info,消息内容：" + message);
    17.  
    18.         message = "warning：Hello World!";
    19.         channel.basicPublish(EXCHANGE_NAME, "warning", null, message.getBytes("UTF-8"));
    20.         System.out.println("发送了消息，" + "等级为warning,消息内容：" + message);
    21.  
    22.         message = "error：Hello World!";
    23.         channel.basicPublish(EXCHANGE_NAME, "error", null, message.getBytes("UTF-8"));
    24.         System.out.println("发送了消息，" + "等级为error,消息内容：" + message);
    25.         channel.close();
    26.         connection.close();
    27.     }
    28. }

    接收所有消息类：ReceiveLogsDirect1.java

    1. public class ReceiveLogsDirect1 {
    2.     private static final String EXCHANGE_NAME = "direct_logs";
    3.  
    4.     public static void main(String[] args) throws IOException, TimeoutException {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         factory.setHost("192.168.119.129");
    7.         factory.setUsername("admin");
    8.         factory.setPassword("password");
    9.         Connection connection = factory.newConnection();
    10.         Channel channel = connection.createChannel();
    11.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    12.         //生成一个随机的临时的queue
    13.         String queueName = channel.queueDeclare().getQueue();
    14.         //一个交换机同时绑定3个queue
    15.         channel.queueBind(queueName, EXCHANGE_NAME, "info");
    16.         channel.queueBind(queueName, EXCHANGE_NAME, "warning");
    17.         channel.queueBind(queueName, EXCHANGE_NAME, "error");
    18.  
    19.         System.out.println("开始接收消息");
    20.         Consumer consumer = new DefaultConsumer(channel) {
    21.             @Override
    22.             public void handleDelivery(String consumerTag, Envelope envelope,
    23.                     BasicProperties properties, byte[] body) throws IOException {
    24.                 String message = new String(body, "UTF-8");
    25.                 System.out.println("收到消息：" + message);
    26.             }
    27.         };
    28.         channel.basicConsume(queueName, true, consumer);
    29.     }
    30. }

    仅仅接收error消息类：ReceiveLogsDirect2.java

    1. public class ReceiveLogsDirect2 {
    2.     private static final String EXCHANGE_NAME = "direct_logs";
    3.  
    4.     public static void main(String[] args) throws IOException, TimeoutException {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         factory.setHost("192.168.119.129");
    7.         factory.setUsername("admin");
    8.         factory.setPassword("password");
    9.         Connection connection = factory.newConnection();
    10.         Channel channel = connection.createChannel();
    11.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    12.         //生成一个随机的临时的queue
    13.         String queueName = channel.queueDeclare().getQueue();
    14.         //一个交换机绑定1个queue
    15.         channel.queueBind(queueName, EXCHANGE_NAME, "error");
    16.  
    17.         System.out.println("开始接收消息");
    18.         Consumer consumer = new DefaultConsumer(channel) {
    19.             @Override
    20.             public void handleDelivery(String consumerTag, Envelope envelope,
    21.                     BasicProperties properties, byte[] body) throws IOException {
    22.                 String message = new String(body, "UTF-8");
    23.                 System.out.println("收到消息：" + message);
    24.             }
    25.         };
    26.         channel.basicConsume(queueName, true, consumer);
    27.     }
    28. }

    运行结果：

     

  • 相关阅读:
    Docker 从构建开始导出一个镜像
    POI导出excel文件之取消合并单元格、删除列、移动列
    vue父子组件传值：父传子、子传父
    【路径最全用法】python代码讲解os.path包的最全用法
    基于java毕业生能力调查评价系统计算机毕业设计源码+系统+lw文档+mysql数据库+调试部署
    【微服务解耦之事件启动】Spring Boot 解耦之事件驱动
    分库分表MySQL
    (十二)判断Python中变量是否是函数
    【密码学】DES 介绍
    Spring Security入门教程，springboot整合Spring Security
  • 原文地址：https://blog.csdn.net/yueyue763184/article/details/127915723