devbox
秋刀鱼在做梦
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ_mq的监视器

作者：秋刀鱼在做梦 | 2024-07-28 02:59:06

赞

踩

mq的监视器

1、什么是MQ

问题1：

系统的注册功能：注册到数据库中、发邮件、发短信

如果串行执行的话，需要耗时3秒

改进： 注册数据库成功，即可，

注册成功后，并行进行发邮件、发短信 ，仅需1秒

Redis数据缓存，它只供获取数据使用，不能发送数据，所以不能使用redis

问题2：

大量的写数据库的请求，如果请求直接到达数据库，将会把数据库写死？

改进：把写数据的请求都写到MQ中，消息的消费者按照队列的顺序逐一取出写入的数据，写入数据库中

流量削峰

某一时间段会有大量的数据发送，达到一定的峰值，针对于这种情况，我们就将消除峰值的操作，称之为，流量削峰

微服务之间的通讯方式：

1、调用

2、MQ

市面上比较火爆的几款MQ：

ActiveMQ，RocketMQ，Kafka，RabbitMQ。

  • 语言的支持：ActiveMQ，RocketMQ只支持Java语言，Kafka可以支持多们语言，RabbitMQ支持多种语言。

  • 效率方面：ActiveMQ，RocketMQ，Kafka效率都是毫秒级别，RabbitMQ是微秒级别的。

  • 消息丢失，消息重复问题： RabbitMQ针对消息的持久化，和重复问题都有比较成熟的解决方案。

  • 学习成本：RabbitMQ非常简单。

RabbitMQ是由Rabbit公司去研发和维护的，最终是在Pivotal。

RabbitMQ严格的遵循AMQP协议，高级消息队列协议，帮助我们在进程之间传递异步消息。

2、搭建环境

1、安装erang依赖环境，一路下一步即可

2、配置环境变量

新建环境变量ERLANG_HOME

编辑系统环境变量Path

3、安装rabbitmq软件，一路下一步即可

4、配置rabbitmq的环境变量

新建环境变量

5、启动RabbitMQ的监视器：

6、启动rabbitmq的服务

7、url地址中访问 localhost:15672

8、账号密码均为mgest

3、RabbitMQ的工作原理

  • Publisher - 生产者：发布消息到RabbitMQ中的Exchange

  • Consumer - 消费者：监听RabbitMQ中的Queue中的消息

  • Exchange - 交换机：和生产者建立连接并接收生产者的消息

  • Queue - 队列：Exchange会将消息分发到指定的Queue，Queue和消费者进行交互

  • Routes - 路由：交换机以什么样的策略将消息发布到Queue

4、RabbitMQ的通讯方式

Hello-World通讯方式

特点：生产环境无法使用

一个生产者、一个消费者、一个交换机、一个队列

1、新建生产者微服务

2、添加依赖

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.springframework.boot</groupId>
  4.         <artifactId>spring-boot-starter-web</artifactId>
  5.     </dependency>
  6.     <dependency>
  7.         <groupId>org.springframework.cloud</groupId>
  8.         <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
  9.     </dependency>
  10.     <dependency>
  11.         <groupId>com.rabbitmq</groupId>
  12.         <artifactId>amqp-client</artifactId>
  13.         <version>5.6.0</version>
  14.     </dependency>
  15. </dependencies>

3、创建启动程序

4、编写配置文件

  1. server:
  2.   port: 8091
  3. spring:
  4.   application:
  5.     name: APPLYPUBLISHERSERVER
  6. eureka:
  7.   client:
  8.     service-url:
  9.       defaultZone: http://root:root@localhost:8080/eureka

5、工具类，在pda-common

  1. package com.qf.pdacommon.util;
  2.  
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  5.  
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  8.  
  9. public class RabbitMQUtil {
  10.     //声明连接工厂，static的
  11.     private static ConnectionFactory factory = null;
  12.     //在静态代码块中初始化工厂信息
  13.     static{
  14.         //创建工厂
  15.         factory = new ConnectionFactory();
  16.         //配置连接信息
  17.         factory.setHost("127.0.0.1");
  18.         //设置端口号
  19.         factory.setPort(5672);
  20. //        //以下可以省略，当自建的用户必须做出如下的配置
  21. //        factory.setUsername("zhangsan");  //登陆账号
  22. //        factory.setPassword("abc123");    //密码
  23. //        factory.setVirtualHost("/pda");   //操作目录
  24.     }
  25.  
  26.     //编写获取连接对象的方法
  27.     public static Connection getConnection(){
  28.         Connection connection=null;
  29.         try {
  30.             connection = factory.newConnection();
  31.         } catch (IOException e) {
  32.             e.printStackTrace();
  33.         } catch (TimeoutException e) {
  34.             e.printStackTrace();
  35.         }
  36.         return connection;
  37.     }
  38. }

6、生产者写入消息到队列中

  1. package com.qf.applypublisher.service.impl;
  2.  
  3. import com.qf.applypublisher.service.PublisherService;
  4. import com.qf.pdacommon.util.RabbitMQUtil;
  5. import com.rabbitmq.client.Channel;
  6. import com.rabbitmq.client.Connection;
  7. import org.springframework.stereotype.Service;
  8.  
  9. import java.io.IOException;
  10. import java.util.concurrent.TimeoutException;
  11.  
  12. @Service
  13. public class PublisherServiceImpl implements PublisherService {
  14.  
  15.  
  16.     @Override
  17.     public String addMq(String meg) {
  18.         //在此将消息写入消息队列
  19.         //1、获取RabbitMQ服务的连接
  20.         Connection connection = RabbitMQUtil.getConnection();
  21.         //通过连接对象获取管道
  22.         Channel channel = null;
  23.         String result ="";
  24.  
  25.         try {
  26.             channel= connection.createChannel();
  27.  
  28.             //生产者把消息通过管道，写入队列中
  29.             //参数1：""使用默认的交换机
  30.             //参数2：队列的名字，也可以作为路由规则,【注意】在一个MQ服务中，队列的名字是唯一的
  31.             //参数3：写入消息时，携带的附加信息，如果没有就传null
  32.             //参数4：写入的消息
  33.             channel.basicPublish("","apply",null,meg.getBytes());
  34.             System.out.println("写入消息队列成功！");
  35.             result="success";
  36.         } catch (IOException e) {
  37.             System.out.println("写入消息失败，发生有异常："+e.getMessage());
  38.             result = "failed";
  39.             e.printStackTrace();
  40.         }finally {
  41.             try {
  42.                 channel.close();
  43.                 connection.close();
  44.             } catch (IOException e) {
  45.                 e.printStackTrace();
  46.             } catch (TimeoutException e) {
  47.                 e.printStackTrace();
  48.             }
  49.  
  50.         }
  51.  
  52.         return result;
  53.     }
  54.  
  55.  
  56. }

7、新建消息的消费者微服务

  1. package com.qf.applyconsumer.service.impl;
  2.  
  3. import com.qf.applyconsumer.service.ConsumerService;
  4. import com.qf.pdacommon.util.RabbitMQUtil;
  5. import com.rabbitmq.client.*;
  6. import org.springframework.stereotype.Service;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. @Service
  12. public class ConsumerServiceImpl implements ConsumerService {
  13.     @Override
  14.     public String getMeg() {
  15.         //到队列中获取消息
  16.         //1、获取连接
  17.         Connection connection = RabbitMQUtil.getConnection();
  18.         //2、创建管道
  19.         Channel channel = null;
  20.         String result = "";
  21.         try {
  22.             channel = connection.createChannel();
  23.  
  24.             //定义队列
  25.             //参数1：获取消息队列的名称
  26.             //参数2：是否给队列中的消息进行持久化，true-进行持久化
  27.             //参数3：当消费者的连接connection对象关闭后(消费者宕机后)，是否删除队列中的消息
  28.             //参数4：当队列中没有消费者消费的时候，是否自动删除队列
  29.             //参数5：指定的队列的附加消息
  30.             channel.queueDeclare("apply",true,false,false,null);
  31.             //声明监听：监听会一直运行，只要监听到队列中有消息就去获取消息
  32.             //通过handleDelivery方法获取消息
  33.             DefaultConsumer consumer = new DefaultConsumer(channel){
  34.                 public void handleDelivery(String consumerTeg, Envelope envelope, AMQP.BasicProperties properties,byte[] body){
  35.                     //参数body就是从队列中获取到的消息
  36.                     //service-> dao  调用其他微服务消息处理
  37.                     String meg = new String(body);
  38.                     System.out.println("获取到的消息是" + meg);
  39.                 }
  40.             };
  41.  
  42.             //自动消息确认：从队列里取出新消息后，自动通知队列，可以删除刚才取出的消息  ACK
  43.             //参数1：队列的名字，告知哪个队列在取出消息后，就自动删除多列中的消息
  44.             //参数2：true-取出消息后，队列就自动删除消息
  45.             //参数3：监听对象
  46.             channel.basicConsume("apply",true,consumer);
  47.             System.out.println("消费者的监听程序开始工作......");
  48.             System.in.read();
  49.             result = "success";
  50.         } catch (IOException e) {
  51.             result = "faileld";
  52.             e.printStackTrace();
  53.         }finally {
  54.             try {
  55.                 channel.close();
  56.                 connection.close();
  57.             } catch (IOException e) {
  58.                 e.printStackTrace();
  59.             } catch (TimeoutException e) {
  60.                 e.printStackTrace();
  61.             }
  62.         }
  63.         return result;
  64.     }
  65. }

 

 

Work工作队列通讯方式

特点：一个生产者、一个交换机、一个队列、多个消费者

新建了一个消费者：确保两个消费者从同一队列中获取消息

【注意】一个消息只能被消费一次，有一个消费者进行消费了，另一个就等到消费完了，删除之后才能取下一个

同时定义两个消费者，除了yml文件中的name名字不一样之外，其他地方一样，生产者不用变，队列名相同就可以

  1. package com.qf.applyconsumerwork2.service.impl;
  2.  
  3. import com.qf.applyconsumerwork2.service.ConsumerWorkService;
  4. import com.qf.pdacommon.util.RabbitMQUtil;
  5. import com.rabbitmq.client.*;
  6. import org.springframework.stereotype.Service;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. @Service
  12. public class ConsumerWorkServiceImpl implements ConsumerWorkService {
  13.     @Override
  14.     public String getMeg() {
  15.  
  16.         //获取连接
  17.         Connection connection = RabbitMQUtil.getConnection();
  18.         //创建管道
  19.         Channel channel = null;
  20.         String result = "";
  21.         try {
  22.             channel = connection.createChannel();
  23.             channel.queueDeclare("apply",true,false,false,null);
  24.  
  25.             //指定消费者的消费能力：一个消费者一次只能最多消费一个消息
  26.             channel.basicQos(1);
  27.  
  28.             //定义channel1，接收外部类的channel，因为内部类中 外部类定义的channel不适用
  29.             Channel channel1 = channel;
  30.             DefaultConsumer consumer = new DefaultConsumer(channel){
  31.                 public void handleDelivery(String consumerTeg, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  32.                     String meg = new String(body);
  33.                     System.out.println("获取到的消息是：" + meg);
  34.                     //改为false只是队列不自动删除消息
  35.                     //需要我们手动删除
  36.                     //判断：消息处理成功了，手动通知队列删除消费者的消息
  37.                     channel1.basicAck(envelope.getDeliveryTag(), false);
  38.                 }
  39.             };
  40.             //改成手动确认，将true改成false
  41.             //从队列里取出消息后，队列不删除取出的消息
  42.             channel.basicConsume("apply",false,consumer);
  43.             System.out.println("8094消费者的监听程序开始工作......");
  44.             System.in.read();
  45.             result =  "Success";
  46.  
  47.         } catch (IOException e) {
  48.             result = "failed";
  49.             e.printStackTrace();
  50.         }finally {
  51.             try {
  52.                 channel.close();
  53.                 connection.close();
  54.             } catch (IOException e) {
  55.                 e.printStackTrace();
  56.             } catch (TimeoutException e) {
  57.                 e.printStackTrace();
  58.             }
  59.         }
  60.  
  61.  
  62.         return result;
  63.     }
  64. }

 

 

发布订阅通讯方式 --- 广播模式

特点：一个生产者、一个交换机、多个队列，每个队列对应多个消费者

定义的所有消费者每一个都会接收到消息

生产者

  1.     //广播模式
  2.     @Override
  3.     public String addMqFanOut(String meg) {
  4.         String result = "";
  5.         Connection connection = RabbitMQUtil.getConnection();
  6.         Channel channel = null;
  7.         try {
  8.             channel = connection.createChannel();
  9.             //定义交换机的类型
  10.             channel.exchangeDeclare("exchange2105", BuiltinExchangeType.FANOUT);
  11.             //定义队列
  12.             //参数1：消息队列的名字
  13.             //参数2：交换机的名字,使用对用的交换机写入到队列里
  14.             //路由器规则
  15.             channel.queueBind("queue1","exchange2105","");
  16.             channel.queueBind("queue2","exchange2105","");
  17.  
  18.             //写入到队列中
  19.             channel.basicPublish("exchange2105","",null,meg.getBytes());
  20.             result = "success";
  21.             System.out.println("广播方式消息写入队列成功");
  22.         } catch (IOException e) {
  23.             e.printStackTrace();
  24.         }finally {
  25.             try {
  26.                 channel.close();
  27.                 connection.close();
  28.             } catch (IOException e) {
  29.                 e.printStackTrace();
  30.             } catch (TimeoutException e) {
  31.                 e.printStackTrace();
  32.             }
  33.  
  34.         }
  35.         return result;
  36.     }
  37.  
  38.  
  39. }

消费者queue1

  1. package com.qf.applyconsumerwork.service.impl;
  2.  
  3. import com.qf.applyconsumerwork.service.ConsumerWorkService;
  4. import com.qf.pdacommon.util.RabbitMQUtil;
  5. import com.rabbitmq.client.*;
  6. import org.springframework.stereotype.Service;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. @Service
  12. public class ConsumerWorkServiceImpl implements ConsumerWorkService {
  13.     @Override
  14.     public String getMeg() {
  15.  
  16.         //获取连接
  17.         Connection connection = RabbitMQUtil.getConnection();
  18.         //创建管道
  19.         Channel channel = null;
  20.         String result = "";
  21.         try {
  22.             channel = connection.createChannel();
  23.             //queue1进行修改，和生产者一致
  24.             channel.queueDeclare("queue1",true,false,false,null);
  25.  
  26.             //指定消费者的消费能力：一个消费者一次只能最多消费一个消息
  27.             channel.basicQos(1);
  28.  
  29.             //定义channel1，接收外部类的channel，因为内部类中 外部类定义的channel不适用
  30.             Channel channel1 = channel;
  31.             DefaultConsumer consumer = new DefaultConsumer(channel){
  32.                 public void handleDelivery(String consumerTeg, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  33.                    String meg = new String(body);
  34.                     System.out.println("8093获取到的消息是：" + meg);
  35.                     //改为false只是队列不自动删除消息
  36.                     //需要我们手动删除
  37.                     //判断：消息处理成功了，手动通知队列删除消费者的消息
  38.                     channel1.basicAck(envelope.getDeliveryTag(), false);
  39.                 }
  40.             };
  41.             //改成手动确认，将true改成false
  42.             //从队列里取出消息后，队列不删除取出的消息
  43.             //queue1进行修改，和生产者一致
  44.             channel.basicConsume("queue1",false,consumer);
  45.             System.out.println("消费者的监听程序开始工作......");
  46.             System.in.read();
  47.             result =  "Success";
  48.  
  49.         } catch (IOException e) {
  50.             result = "failed";
  51.             e.printStackTrace();
  52.         }finally {
  53.             try {
  54.                 channel.close();
  55.                 connection.close();
  56.             } catch (IOException e) {
  57.                 e.printStackTrace();
  58.             } catch (TimeoutException e) {
  59.                 e.printStackTrace();
  60.             }
  61.         }
  62.  
  63.  
  64.         return result;
  65.     }
  66. }

消费者queue2

  1. package com.qf.applyconsumerwork2.service.impl;
  2.  
  3. import com.qf.applyconsumerwork2.service.ConsumerWorkService;
  4. import com.qf.pdacommon.util.RabbitMQUtil;
  5. import com.rabbitmq.client.*;
  6. import org.springframework.stereotype.Service;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. @Service
  12. public class ConsumerWorkServiceImpl implements ConsumerWorkService {
  13.     @Override
  14.     public String getMeg() {
  15.  
  16.         //获取连接
  17.         Connection connection = RabbitMQUtil.getConnection();
  18.         //创建管道
  19.         Channel channel = null;
  20.         String result = "";
  21.         try {
  22.             channel = connection.createChannel();
  23.             //queue2进行修改，和生产者一致
  24.             channel.queueDeclare("queue2",true,false,false,null);
  25.  
  26.             //指定消费者的消费能力：一个消费者一次只能最多消费一个消息
  27.             channel.basicQos(1);
  28.  
  29.             //定义channel1，接收外部类的channel，因为内部类中 外部类定义的channel不适用
  30.             Channel channel1 = channel;
  31.             DefaultConsumer consumer = new DefaultConsumer(channel){
  32.                 public void handleDelivery(String consumerTeg, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  33.                     String meg = new String(body);
  34.                     System.out.println("获取到的消息是：" + meg);
  35.                     //改为false只是队列不自动删除消息
  36.                     //需要我们手动删除
  37.                     //判断：消息处理成功了，手动通知队列删除消费者的消息
  38.                     channel1.basicAck(envelope.getDeliveryTag(), false);
  39.                 }
  40.             };
  41.  
  42.             //队列名字是消费者中定义的队列名字
  43.             //queue2进行修改，和生产者一致
  44.             channel.basicConsume("queue2",false,consumer);
  45.             System.out.println("8094消费者的监听程序开始工作......");
  46.             System.in.read();
  47.             result =  "Success";
  48.  
  49.         } catch (IOException e) {
  50.             result = "failed";
  51.             e.printStackTrace();
  52.         }finally {
  53.             try {
  54.                 channel.close();
  55.                 connection.close();
  56.             } catch (IOException e) {
  57.                 e.printStackTrace();
  58.             } catch (TimeoutException e) {
  59.                 e.printStackTrace();
  60.             }
  61.         }
  62.  
  63.  
  64.         return result;
  65.     }
  66. }

 

Routing路由通讯方式

生产者：交换机类型是Direct

               定义队列时指定了路由规则

生产者

  1.     //路由方式
  2.     @Override
  3.     public String addMqDirect(String meg, String route) {
  4.         Connection connection = RabbitMQUtil.getConnection();
  5.         Channel channel = null;
  6.         String result = "";
  7.         try {
  8.             channel = connection.createChannel();
  9.  
  10.             //定义交换机的类型
  11.             channel.exchangeDeclare("exchange210502",BuiltinExchangeType.DIRECT);
  12.             //绑定队列，同时指定路由规则
  13.             channel.queueBind("queue3","exchange210502","goods");
  14.             channel.queueBind("queue4","exchange210502","score");
  15.  
  16.             //写入消息
  17.             channel.basicPublish("exchange210502",route,null,meg.getBytes());
  18.             System.out.println("Direct消息写入成功");
  19.             result = "success";
  20.         } catch (IOException e) {
  21.             e.printStackTrace();
  22.         }
  23.         return result;
  24.     }

消费者queue3  ---- queue4中把3改为4

  1. package com.qf.applyconsumerwork.service.impl;
  2.  
  3. import com.qf.applyconsumerwork.service.ConsumerWorkService;
  4. import com.qf.pdacommon.util.RabbitMQUtil;
  5. import com.rabbitmq.client.*;
  6. import org.springframework.stereotype.Service;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. @Service
  12. public class ConsumerWorkServiceImpl implements ConsumerWorkService {
  13.     @Override
  14.     public String getMeg() {
  15.  
  16.         //获取连接
  17.         Connection connection = RabbitMQUtil.getConnection();
  18.         //创建管道
  19.         Channel channel = null;
  20.         String result = "";
  21.         try {
  22.             channel = connection.createChannel();
  23.             channel.queueDeclare("queue3",true,false,false,null);
  24.  
  25.             //指定消费者的消费能力：一个消费者一次只能最多消费一个消息
  26.             channel.basicQos(1);
  27.  
  28.             //定义channel1，接收外部类的channel，因为内部类中 外部类定义的channel不适用
  29.             Channel channel1 = channel;
  30.             DefaultConsumer consumer = new DefaultConsumer(channel){
  31.                 public void handleDelivery(String consumerTeg, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  32.                    String meg = new String(body);
  33.                     System.out.println("8093获取到的消息是：" + meg);
  34.                     //改为false只是队列不自动删除消息
  35.                     //需要我们手动删除
  36.                     //判断：消息处理成功了，手动通知队列删除消费者的消息
  37.                     channel1.basicAck(envelope.getDeliveryTag(), false);
  38.                 }
  39.             };
  40.             //改成手动确认，将true改成false
  41.             //从队列里取出消息后，队列不删除取出的消息
  42.             channel.basicConsume("queue3",false,consumer);
  43.             System.out.println("8093消费者的监听程序开始工作......");
  44.             System.in.read();
  45.             result =  "Success";
  46.  
  47.         } catch (IOException e) {
  48.             result = "failed";
  49.             e.printStackTrace();
  50.         }finally {
  51.             try {
  52.                 channel.close();
  53.                 connection.close();
  54.             } catch (IOException e) {
  55.                 e.printStackTrace();
  56.             } catch (TimeoutException e) {
  57.                 e.printStackTrace();
  58.             }
  59.         }
  60.  
  61.  
  62.         return result;
  63.     }
  64. }

Topic通讯方式


 

生产者

  1.     @Override
  2.     public String addTopic(String meg, String route) {
  3.         Connection connection = RabbitMQUtil.getConnection();
  4.         Channel channel = null;
  5.         String result = "";
  6.  
  7.         try {
  8.             channel = connection.createChannel();
  9.  
  10.             //定义交换机的类型
  11.             channel.exchangeDeclare("exchange210504",BuiltinExchangeType.TOPIC);
  12.  
  13.             //绑定队列：交换机按照什么路由规则，写入哪个队列
  14.             channel.queueBind("queue5","exchange210504","liaoning.dalian.*");
  15.             channel.queueBind("queue6","exchange210504","heilongjiang.#");
  16.  
  17.             //写入消息
  18.             channel.basicPublish("exchange210504",route,null,meg.getBytes());
  19.             System.out.println("Topic消息写入成功");
  20.             result = "success";
  21.         } catch (IOException e) {
  22.             e.printStackTrace();
  23.         }
  24.         return result;
  25.     }

消费者

  1. package com.qf.applyconsumerwork.service.impl;
  2.  
  3. import com.qf.applyconsumerwork.service.ConsumerWorkService;
  4. import com.qf.pdacommon.util.RabbitMQUtil;
  5. import com.rabbitmq.client.*;
  6. import org.springframework.stereotype.Service;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. @Service
  12. public class ConsumerWorkServiceImpl implements ConsumerWorkService {
  13.     @Override
  14.     public String getMeg() {
  15.  
  16.         //获取连接
  17.         Connection connection = RabbitMQUtil.getConnection();
  18.         //创建管道
  19.         Channel channel = null;
  20.         String result = "";
  21.         try {
  22.             channel = connection.createChannel();
  23.             channel.queueDeclare("queue5",true,false,false,null);
  24.  
  25.             //指定消费者的消费能力：一个消费者一次只能最多消费一个消息
  26.             channel.basicQos(1);
  27.  
  28.             //定义channel1，接收外部类的channel，因为内部类中 外部类定义的channel不适用
  29.             Channel channel1 = channel;
  30.             DefaultConsumer consumer = new DefaultConsumer(channel){
  31.                 public void handleDelivery(String consumerTeg, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  32.                    String meg = new String(body);
  33.                     System.out.println("8093获取到的消息是：" + meg);
  34.                     //改为false只是队列不自动删除消息
  35.                     //需要我们手动删除
  36.                     //判断：消息处理成功了，手动通知队列删除消费者的消息
  37.                     channel1.basicAck(envelope.getDeliveryTag(), false);
  38.                 }
  39.             };
  40.             //改成手动确认，将true改成false
  41.             //从队列里取出消息后，队列不删除取出的消息
  42.             channel.basicConsume("queue5",false,consumer);
  43.             System.out.println("8093消费者的监听程序开始工作......");
  44.             System.in.read();
  45.             result =  "success";
  46.  
  47.         } catch (IOException e) {
  48.             result = "failed";
  49.             e.printStackTrace();
  50.         }finally {
  51.             try {
  52.                 channel.close();
  53.                 connection.close();
  54.             } catch (IOException e) {
  55.                 e.printStackTrace();
  56.             } catch (TimeoutException e) {
  57.                 e.printStackTrace();
  58.             }
  59.         }
  60.  
  61.  
  62.         return result;
  63.     }
  64. }

如果没有问题但是报错就看一下http://localhost:15672/#/exchanges

 更改交换机的名字

5、消息的可靠性---重要

从如下方面考虑可靠性：

1、生产者要把消息成功的写入交换机 ----- confirm机制

2、交换机要成功的把消息写入队列

3、队列要可靠 ---- 本身具有的持久化机制

4、确保消费者成功消费消息后，才把消息从队列中删除 --- 手动ACK

Confirm 普通消息确认机制

  1. @Override
  2. public String addMqTopic(String meg, String route) {
  3.     String result="";
  4.     Connection connection=RabbitMQUtil.getConnection();
  5.     Channel channel=null;
  6.     try {
  7.         channel=connection.createChannel();
  8.         //开启Confirm机制---确保生产者把消息成功的写入到交换机中
  9.         channel.confirmSelect();
  10.         //定义交换机
  11.         channel.exchangeDeclare("exchange210503",BuiltinExchangeType.TOPIC);
  12.  
  13.         //绑定队列：交换机按照什么路由规则，写入哪个队列
  14.         channel.queueBind("queue5","exchange210503","liaoning.dalian.*");
  15.         channel.queueBind("queue6","exchange210503","heilongjiang.#");
  16.  
  17.         //写入队列
  18.         channel.basicPublish("exchange210503",route,null,meg.getBytes());
  19.  
  20.         //判断消息写入交换机是否成功!
  21.         if(channel.waitForConfirms()){
  22.             result="写入交换机成功！";
  23.         }else {
  24.             result="写入交换机失败!";
  25.         }
  26.     } catch (IOException | InterruptedException e) {
  27.         e.printStackTrace();
  28.     } finally {
  29.         try {
  30.             channel.close();
  31.             connection.close();
  32.         } catch (IOException e) {
  33.             e.printStackTrace();
  34.         } catch (TimeoutException e) {
  35.             e.printStackTrace();
  36.         }
  37.     }
  38.     return result;
  39. }

批量消息确认机制（Confirm）

  1. @Override
  2. public String addMq(String meg) {
  3.     //在此处将消息写入消息队列
  4.     //1、获取RabbitMQ服务的连接
  5.     Connection connection= RabbitMQUtil.getConnection();
  6.     //2、通过连接对象获取管道
  7.     Channel channel=null;
  8.     String reslult="";
  9.     try {
  10.         channel=connection.createChannel();
  11.         //开启消息确认机制
  12.         channel.confirmSelect();
  13.         //批量写入消息
  14.         //生产者把消息通过管道，写入队列中
  15.         //参数1：""使用默认的交换机
  16.         //参数2：队列的名字,也可以作为路由规则，【注意】在一个MQ服务中，队列的名字是唯一的
  17.         //参数3：写入消息时，携带的附加信息，如果没有就传null
  18.         //参数4：写入的消息
  19.         for (int i=1; i<=10; i++){
  20.             meg=meg+i;
  21.             channel.basicPublish("","apply",null,meg.getBytes());
  22.         }
  23.         //批量确认，如果有一个消息写入交换机失败，则之前写过的所有消息都失败，失败后会抛出异常
  24.         channel.waitForConfirmsOrDie();
  25.         System.out.println("写入消息队列成功！");
  26.         reslult="批量将消息写入交换机成功！";
  27.     } catch (IOException | InterruptedException e) {
  28.         System.out.println("写入消息失败，发生有异常："+e.getMessage());
  29.         reslult="批量写入交换机失败！";
  30.         e.printStackTrace();
  31.     }finally {
  32.         try {
  33.             channel.close();
  34.             connection.close();
  35.         } catch (IOException e) {
  36.             e.printStackTrace();
  37.         } catch (TimeoutException e) {
  38.             e.printStackTrace();
  39.         }
  40.     }
  41.     return reslult;
  42. }

异步Confirm消息确认

  1. @Override
  2. public String addMqFanOut(String meg){
  3.     String result="";
  4.     Connection connection=RabbitMQUtil.getConnection();
  5.     Channel channel=null;
  6.     try {
  7.         channel=connection.createChannel();
  8.         //开启Confirm消息确认机制
  9.         channel.confirmSelect();
  10.         //定义交换机的类型
  11.         channel.exchangeDeclare("exchange2105", BuiltinExchangeType.FANOUT);
  12.         //定义队列
  13.         //参数1：队列名字，不能重复
  14.         //参数2：交换机名字，使用对应没名字的交换机写入到该队列中
  15.         channel.queueBind("queue1","exchange2105","");
  16.         channel.queueBind("queue2","exchange2105","");
  17.  
  18.         //写入到队列中
  19.         channel.basicPublish("exchange2105","",null,meg.getBytes());
  20.         //异步消息确认
  21.         channel.addConfirmListener(new ConfirmListener() {
  22.             @Override
  23.             public void handleAck(long l, boolean b) throws IOException {
  24.                 //消息写入成功时，会调用
  25.                 System.out.println("写入成功！消息的标识："+l+"，是否为批量？"+b);
  26.             }
  27.  
  28.             @Override
  29.             public void handleNack(long l, boolean b) throws IOException {
  30.                 //消息写入失败时回调
  31.                 System.out.println("写入失败！消息的标识："+l+"，是否为批量？"+b);
  32.  
  33.             }
  34.         });
  35.  
  36.         result="success";
  37.         System.out.println("发布、订阅方式将消息写入队列成功！");
  38.  
  39.     } catch (IOException e) {
  40.         e.printStackTrace();
  41.     } finally {
  42.         try {
  43.             channel.close();
  44.             connection.close();
  45.         } catch (IOException e) {
  46.             e.printStackTrace();
  47.         } catch (TimeoutException e) {
  48.             e.printStackTrace();
  49.         }
  50.     }
  51.     return null;
  52. }

Return机制---确保消息从交换机写入队列成功

6、避免消息的重复消费

产生的原因：没有及时 进行手动ACK

解决思路：

1、在Redis中设置一个值 set mqKey 0 0---没有程序去消费这个消息

2、消费者在到队列中取消息之前，先读取mqKey

如果读取到的是0，则允许读取消息，否则不允许

3、在读取过程中，设置为1 1---此消息正在被某个消费者消费        

当一个消费者进入消息队列之后，将其key值进行改变，正常情况之下，在执行完流程，将信息存储进数据库，然后正常出队列之后进行删除，将key改回去，但是有一种情况，在没有执行完就离开队列key没有改回去，这种情况下，谁都进不来，产生了死锁现象

解决的办法就是设置有效时间

  1. public void getMeg() {
  2.     //到队列中获取消息
  3.     //1、获取连接
  4.     Connection connection= RabbitMQUtil.getConnection();
  5.     //2、创建管道
  6.     Channel channel=null;
  7.     String result="";
  8.     try {
  9.         channel=connection.createChannel();
  10.  
  11.         //定义队列
  12.         //参数1：获取消息的队列的名称
  13.         //参数2：是否给队列中的消息进行持久化  true-进行持久化
  14.         //参数3：当消费者的连接connection对象关闭后(消费者宕机后)，是否删除队列中的消息
  15.         //参数4：当队列中没有消费者消费的时候，是否自动删除队列
  16.         //参数5：指定的队列的附加消息
  17.         channel.queueDeclare("queue5",true,false,false,null);
  18.  
  19.         //指定消费者的消费能力：一个消费者一次只能最多消费一个消息
  20.         channel.basicQos(1);
  21.  
  22.         //声明监听：监听会一直运行，只要监听到队列中有消息，就去获取消息
  23.         //通过handleDelivery方法获取消息
  24.         Channel channel1=channel;
  25.         DefaultConsumer consumer=new DefaultConsumer(channel1){
  26.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  27.                 //获取Redis的连接
  28.                 Jedis jedis= JedisUtil.getJedis();
  29.                 if(jedis!=null){
  30.                     jedis.select(5);
  31.                     //读取锁里的信息
  32.                     String redisLock=jedis.get("mqKey");
  33.                     if(redisLock==null){
  34.                         //设置只当时间，时间到自动将key改回去，赋值
  35.                         jedis.setex("mqKey",10,"1");
  36.                         //参数body就是从队列中获取到的消息
  37.                         //service->dao   调用其他微服务处理消息
  38.                         String meg=new String(body);
  39.                         System.out.println("1号消费者获取到消息："+meg);
  40.                         //判断，消息处理成功了，手动通知队列删除消费的消息
  41.                         channel1.basicAck(envelope.getDeliveryTag(),false);
  42.                         //消费完成之后，设置为0
  43.                         jedis.set("mqKey","0");
  44.                     }else{
  45.                         int intRedisLock=Integer.parseInt(redisLock);
  46.                         if(intRedisLock==0){
  47.                             //赋值
  48.                             jedis.setex("mqKey",10,"1");
  49.                             //参数body就是从队列中获取到的消息
  50.                             //service->dao   调用其他微服务处理消息
  51.                             String meg=new String(body);
  52.                             System.out.println("1号消费者获取到消息："+meg);
  53.                             //判断，消息处理成功了，手动通知队列删除消费的消息
  54.                             channel1.basicAck(envelope.getDeliveryTag(),false);
  55.                             //消费完成之后，设置为0
  56.                             jedis.set("mqKey","0");
  57.                         }
  58.                     }
  59.                 }
  60.             }
  61.         };

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

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

  

闽ICP备14008679号