devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

消息队列之RabbitMQ的五种消息模型，及如何保证可靠消息最终一致性_rabbitmq五种消息模型

作者：小小林熬夜学编程 | 2024-02-21 22:35:27

赞

踩

rabbitmq五种消息模型

什么是MQ

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

 MQ的选型和对比

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

其架构如图：

 从这个图可以了解到RabbitMQ的四个重要接口：

connection：连接
channel：轻量级的connection，信道，核心处理，大部分的API实现
exchange：只负责分发消息，若没有queue绑定到exchange上则消息会丢失
queue：存储消息的容器

 2007年，Rabbit技术公司基于AMQP标准开发的RabbitMQ1.0发布。RabbitMQ采用Erlang 语言开发。

RabbitMQ

使用RabbitMQ的三大好处：

解耦

传统系统之间的耦合度太强，主要集中在两部分：

业务之间的耦合度和系统之间的耦合度。

比如说存在这样一个微服务场景，A模块负责图书资源，B模块负责搜索，此时，A模块新增了图书资源，B模块需要同步索引库，那同步索引库要如何操作？直接放在A模块的业务逻辑中去实现？这不符合微服务的理念，A模块既要管理数据库又要管理索引库显然是不行的。那通过feign去调用呢？显然这样可以实现业务逻辑，但系统之间的耦合度又上来了，如果后面要扩展一个C模块，同样要在A模块新增的时候执行业务，那A模块就得改代码。

而使用了RabbitMQ之后，可以将消息写入消息队列，需要消息的系统自己从消息队列中订阅即可，同样是上面那个场景，A新增的同时，将新增信息存入消息队列，B模块需要这个消息从消息队列直接订阅，就算后面扩展了C模块，同样也从消息队列中订阅即可，A模块无需改动任何代码。

可以参考一下图片：

  

2.异步

将消息写入消息队列，非必要的业务逻辑以异步方式运行，加快响应速度。

如：假设存在一下的业务执行流程，用户下订单，成功后发送短信通知，发送邮件通知，并在app推送通知。这个过程中用户执行完下订单操作后，订单服务耗时50ms，然后订单服务以异步的方式将消息存入消息队列，然后返回给用户响应信息，短信服务，邮件服务和app推送服务，短信服务，邮件服务，app推送服务耗时都是50ms，这样整个业务执行的耗时就是50ms，而如果使用传统方式去执行，耗时200ms。

3.削峰 

削峰是解决了并发问题，在传统模式下，并发量太大的时候，所有的请求直接怼到数据库，造成数据库连接异常。加入消息队列后，系统可以根据数据库能处理的并发量，分批次从数据库拉取数据。

RabbitMQ的五种消息模型

（提一下，RabbitMQ官方提供的有6中消息模型，只是第六种属于RPC并不是MQ，所以这里没有将其列入）

Simple-简单模型：

在这个模型下，RabbitMQ是一个消息代理：它接受和转发消息。可以把他想象成一个邮政信箱。

RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。

 

P（producer/ publisher）：生产者，一个发送消息的用户应用程序。

C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序

队列（红色区域）：rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。

总之：

生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。

在这个模式下，一个生产者对应一个消费者

代码实现如下：

生产者

  1. import com.rabbitmq.client.Channel;
  2. import com.rabbitmq.client.Connection;
  3. import com.alex_hh.simple.util.ConnectionUtil;
  4.  
  5. public class Send {
  6.  
  7.     private final static String QUEUE_NAME = "simple_queue";
  8.  
  9.     public static void main(String[] argv) throws Exception {
  10.         // 获取到连接以及mq通道
  11.         Connection connection = ConnectionUtil.getConnection();
  12.         // 轻量级的 Connection，这是完成大部分API的地方。
  13.         Channel channel = connection.createChannel();
  14.  
  15.         // 声明（创建）队列，必须声明队列才能够发送消息，我们可以把消息发送到队列中。
  16.         // 声明一个队列是幂等的 - 只有当它不存在时才会被创建
  17.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  18.  
  19.         // 消息内容
  20.         String message = "Hello World!";
  21.         channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
  22.         System.out.println(" [x] Sent '" + message + "'");
  23.  
  24.         //关闭通道和连接
  25.         channel.close();
  26.         connection.close();
  27.     }
  28. }

消费者：

  1. import java.io.IOException;
  2. import com.rabbitmq.client.AMQP.BasicProperties;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.DefaultConsumer;
  6. import com.rabbitmq.client.Envelope;
  7. import com.alex_hh.simple.util.ConnectionUtil;
  8.  
  9. public class Recv {
  10.     private final static String QUEUE_NAME = "simple_queue";
  11.  
  12.     public static void main(String[] argv) throws Exception {
  13.         // 获取到连接
  14.         Connection connection = ConnectionUtil.getConnection();
  15.         // 创建通道
  16.         Channel channel = connection.createChannel();
  17.         // 声明队列
  18.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  19.         // 定义队列的消费者
  20.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  21.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  22.             @Override
  23.             public void handleDelivery(String consumerTag, Envelope envelope, 
  24.                           BasicProperties properties,byte[] body) throws IOException {
  25.                 // body 即消息体
  26.                 String msg = new String(body);
  27.                 System.out.println(" [消费者1] received : " + msg + "!");
  28.             }
  29.         };
  30.         // 监听队列，第二个参数：是否自动进行消息确认。
  31.         channel.basicConsume(QUEUE_NAME, true, consumer);
  32.     }
  33. }

消息确定机制（ACK）：

消息一旦被消费者接收，队列中的消息就会被删除。

那么问题来了：RabbitMQ怎么知道消息被接收了呢？

如果消费者领取消息后，还没执行操作就挂掉了呢？或者抛出了异常？消息消费失败，但是RabbitMQ无从得知，这样消息就丢失了！

因此，RabbitMQ有一个ACK机制。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：

  • 自动ACK：消息一旦被接收，消费者自动发送ACK

  • 手动ACK：消息接收后，不会发送ACK，需要手动调用

什么使用自动ACK，什么时候使用手动ACK

这需要看消息的重要性：

  • 如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便

  • 如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK，RabbitMQ就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。

开启手动ACK

 修改消费者代码：

  1. public static void main(String[] argv) throws Exception {
  2.         // 获取到连接
  3.         Connection connection = ConnectionUtil.getConnection();
  4.         // 创建通道
  5.         Channel channel = connection.createChannel();
  6.         // 声明队列
  7.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  8.         // 定义队列的消费者
  9.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  10.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  11.             @Override
  12.             public void handleDelivery(String consumerTag, Envelope envelope, 
  13.                           BasicProperties properties,byte[] body) throws IOException {
  14.                 // body 即消息体
  15.                 String msg = new String(body);
  16.                 System.out.println(" [消费者1] received : " + msg + "!");
  17.                 //开启手动ACK
  18.                 channel.basicAck(envelope.getDeliveryTag(),false);
  19.             }
  20.         };
  21.         // 监听队列，第二个参数：是否自动进行消息确认。
  22.         channel.basicConsume(QUEUE_NAME, false, consumer);
  23.     }

Work-工作模型

 工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多消费者时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。

这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。

在上面的基础上，做些改进，调整生产者生产多条消息

  1. public class Send {
  2.     private final static String QUEUE_NAME = "test_work_queue";
  3.  
  4.     public static void main(String[] argv) throws Exception {
  5.         // 获取到连接
  6.         Connection connection = ConnectionUtil.getConnection();
  7.         // 获取通道
  8.         Channel channel = connection.createChannel();
  9.         // 声明队列
  10.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  11.         // 循环发布任务
  12.         for (int i = 0; i < 50; i++) {
  13.             // 消息内容
  14.             String message = "task .. " + i;
  15.             channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
  16.             System.out.println(" [x] Sent '" + message + "'");
  17.  
  18.             Thread.sleep(i * 2);
  19.         }
  20.         // 关闭通道和连接
  21.         channel.close();
  22.         connection.close();
  23.     }
  24. }

创建一个新的消费者2

  1. public static void main(String[] argv) throws Exception {
  2.         // 获取到连接
  3.         Connection connection = ConnectionUtil.getConnection();
  4.         // 创建通道
  5.         Channel channel = connection.createChannel();
  6.         // 声明队列
  7.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  8.         // 定义队列的消费者
  9.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  10.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  11.             @Override
  12.             public void handleDelivery(String consumerTag, Envelope envelope, 
  13.                           BasicProperties properties,byte[] body) throws IOException {
  14.                 // body 即消息体
  15.                 String msg = new String(body);
  16.                 System.out.println(" [消费者2] received : " + msg + "!");
  17.                 try{
  18.                     //模拟任务完成耗时
  19.                     Thread.sleep(1000);
  20.                 }catch(Exception e){
  21.                     
  22.                 }
  23.                 //开启手动ACK
  24.                 channel.basicAck(envelope.getDeliveryTag(),false);
  25.             }
  26.         };
  27.         // 监听队列，第二个参数：是否自动进行消息确认。
  28.         channel.basicConsume(QUEUE_NAME, false, consumer);
  29.     }

两个消费者一同启动，然后生产者发送50条消息，可以发现，两个消费者各自消费了25条消息，而且各不相同，这就实现了任务的分发。

 

但是其中存在问题：

  • 消费者1比消费者2的效率要低，一次任务的耗时较长

  • 然而两人最终消费的消息数量是一样的

  • 消费者2大量时间处于空闲状态，消费者1一直忙碌

现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。

怎么实现呢？

我们可以使用basicQos方法和prefetchCount = 1设置。 这告诉RabbitMQ一次不要向工作人员发送多于一条消息。 或者换句话说，不要向工作人员发送新消息，直到它处理并确认了前一个消息。 相反，它会将其分派给不是仍然忙碌的下一个工作人员。

在消费者上加入：

channel.basicQos(1);

 再次执行：

 Fanout-广播模型：

在广播模式下，消息发送流程是这样的：

  • 1） 可以有多个消费者

  • 2） 每个消费者有自己的queue（队列）

  • 3） 每个队列都要绑定到Exchange（交换机）

  • 4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。

  • 5） 交换机把消息发送给绑定过的所有队列

  • 6） 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

 

 在广播模式下，生产者不在将消息发送至队列，而是发送给交换机，由交换机来实现消息的分配，改一下生产者的代码

  1. public class Send {
  2.  
  3.     private final static String EXCHANGE_NAME = "fanout_exchange_test";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         // 获取到连接
  7.         Connection connection = ConnectionUtil.getConnection();
  8.         // 获取通道
  9.         Channel channel = connection.createChannel();
  10.         
  11.         // 声明exchange，指定类型为fanout
  12.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
  13.         
  14.         // 消息内容
  15.         String message = "Hello everyone";
  16.         // 发布消息到Exchange
  17.         channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
  18.         System.out.println(" [生产者] Sent '" + message + "'");
  19.  
  20.         channel.close();
  21.         connection.close();
  22.     }
  23. }

消费者需要声明队列并绑定交换机，更改代码如下：

消费者1

  1. public class Recv {
  2.     private final static String QUEUE_NAME = "fanout_exchange_queue_1";
  3.  
  4.     private final static String EXCHANGE_NAME = "fanout_exchange_test";
  5.  
  6.     public static void main(String[] argv) throws Exception {
  7.         // 获取到连接
  8.         Connection connection = ConnectionUtil.getConnection();
  9.         // 获取通道
  10.         Channel channel = connection.createChannel();
  11.         // 声明队列
  12.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  13.  
  14.         // 绑定队列到交换机
  15.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
  16.  
  17.         // 定义队列的消费者
  18.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  19.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  20.             @Override
  21.             public void handleDelivery(String consumerTag, Envelope envelope, 
  22.                           BasicProperties properties,byte[] body) throws IOException {
  23.                 // body 即消息体
  24.                 String msg = new String(body);
  25.                 System.out.println(" [消费者1] received : " + msg + "!");
  26.             }
  27.         };
  28.         // 监听队列，自动返回完成
  29.         channel.basicConsume(QUEUE_NAME, true, consumer);
  30.     }
  31. }

消费者2

  1. public class Recv2 {
  2.     private final static String QUEUE_NAME = "fanout_exchange_queue_2";
  3.  
  4.     private final static String EXCHANGE_NAME = "fanout_exchange_test";
  5.  
  6.     public static void main(String[] argv) throws Exception {
  7.         // 获取到连接
  8.         Connection connection = ConnectionUtil.getConnection();
  9.         // 获取通道
  10.         Channel channel = connection.createChannel();
  11.         // 声明队列
  12.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  13.  
  14.         // 绑定队列到交换机
  15.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
  16.         
  17.         // 定义队列的消费者
  18.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  19.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  20.             @Override
  21.             public void handleDelivery(String consumerTag, Envelope envelope, 
  22.                           BasicProperties properties,byte[] body) throws IOException {
  23.                 // body 即消息体
  24.                 String msg = new String(body);
  25.                 System.out.println(" [消费者2] received : " + msg + "!");
  26.             }
  27.         };
  28.          // 监听队列，自动返回完成
  29.         channel.basicConsume(QUEUE_NAME, true, consumer);
  30.     }
  31. }

 测试：运行两个消费者，并启动生产者发送一条消息

Direct-定向模型：

有选择性的接收消息

在订阅模式（广播）中，生产者发布消息，所有消费者都可以获取所有消息。

在路由模式（定向）中，我们将添加一个功能 - 我们将只能订阅一部分消息。 例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。

但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）

消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。

 

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。

X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列

C1：消费者，其所在队列指定了需要routing key 为 error 的消息

C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

 与前面基本相似，在生产者声明交换机时指定类型为：“direct”

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

 发送消息时，指定routing key

channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());

 消费者绑定交换机时指定routingkey

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");

 Topic-主题模型：

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

如：

audit.#：能够匹配audit.irs.corporate 或者 audit.irs

audit.*：只能匹配audit.irs

 使用与Direct模式基本一样，只是生产者在声明交换机的时候指定类型为：‘topic’

channel.exchangeDeclare(EXCHANGE_NAME, "topic"); 

 消息的持久化：

如何避免消息丢失？

1） 消费者的手动ACK机制。可以防止业务处理失败。

2） 但是，如果在消费者消费之前，MQ就宕机了，消息就没了。

是可以将消息进行持久化呢？

要将消息持久化，前提是：队列、Exchange都持久化

交换机的持久化

//生产者声明交换机的时候，在第三个属性（durable），配置为true即可

 channel.exchangeDeclare(EXCHANGE_NAME, "topic",true); 

队列的持久化

消费者声明队列的时候，将第二个属性（durable），配置为true即可

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

消息的持久化：

生产者发送消息时，指定第三个参数为

channel.basicPublish(EXCHANGE_NAME, "insert", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

 总结一下五种模型，simple和work都是生产者直接发送消息到消息队列，而simple是一对一的关系而work是一对多的关系

fanout,direct,topic都是通过交换机来发送消息，不同在于fanout发送的消息所有订阅到交换机上的消费者都能获取，direct与topic可以指定routing key来让部分持有routing key的订阅者获取消息，而topic可以指定通配符，direct不行。

 如何实现可靠消息最终一致性

在实际系统的开发过程中，可能服务间的调用是异步的。也就是说，一个服务发送一个消息给 MQ，即消息中间件，比如RocketMQ、RabbitMQ、Kafka、ActiveMQ 等等。

然后，另外一个服务从 MQ 消费到一条消息后进行处理。这就成了基于 MQ 的异步调用了。

那么针对这种基于 MQ 的异步调用，如何保证各个服务间的分布式事务呢？也就是说，我希望的是基于MQ 实现异步调用的多个服务的业务逻辑，要么一起成功，要么一起失败。这个时候，就要用上可靠消息最终一致性方案，来实现分布式事务。

可靠消息最终一致性方案是指当事务发起方执行完成本地事务后并发出一条消息，事务参与方(消息消费者)一定能够接收消息并处理事务成功，此方案强调的是只要消息发给事务参与方最终事务要达到一致。

此方案是利用消息中间件完成，如下图：

事务发起方（消息生产方）将消息发给消息中间件，事务参与方从消息中间件接收消息，事务发起方和消息中间件之间，事务参与方（消息消费方）和消息中间件之间都是通过网络通信，由于网络通信的不确定性会导致分布式事务问题。

可靠消息一致性需要解决的问题：

1.上游服务把消息成功发送：

本地事务与消息发送的原子性问题：事务发起方在本地事务执行成功后消息必须发出去，否则就回滚事务。即实现本地事务和消息发送的原子性，要么都成功，要么都失败。（可靠消息）

2.下游服务把消息成功消费：

事务参与方接收消息的可靠性：事务参与方必须能够从消息队列接收到消息。（可靠消息）

3.对消息做幂等处理：

消息重复消费的问题：由于网络2的存在，若某一个消费节点响应超时但是消费成功，此时消息中间件会重复投递此消息，就导致了消息的重复消费。（最终一致性）

解决方案：

为了让上游服务把消息成功发出，可以使用本地消息表，该方案最初是eBay提出的，在系统A处理任务完成后，在本地记录待发送信息。一个定时任务不断检查，是否发送成功，如果发送成功，将记录状态修改。如下图：

 如处理任务后，在本地消息表中添加一条数据，发送消息到中间件，随后通过异步的方式等待响应，如果响应成功返回，则修改数据的记录状态（也可以直接删除），如果收不到响应，则通过定时任务一直发送，直至成功。

为了让下游服务把消息成功消费，可以使用

消息持久化：可保证消息中间件宕机后消息不丢失

手动ack：保证消息投递失败时消息的重新投递

 实现消息的幂等性，可以通过消息去重表。

消息去重表：任务B处理消息前，先查询该消息是否被消费，如果没消费，处理任务B成功，记录消息。如果消息已经被消费，直接返回应答成功

 代码实现：

创建本地消息记录表：

DROP TABLE IF EXISTS `local_message`;
CREATE TABLE `local_message` (
  `tx_no` varchar(255) NOT NULL,
  `item_id` bigint DEFAULT NULL,
  `state` int(11) DEFAULT NULL,
  PRIMARY KEY (`tx_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

新增信息去重表

DROP TABLE IF EXISTS `msg_distinct`;
CREATE TABLE `msg_distinct` (
  `tx_no` varchar(255) NOT NULL,
  `create_time` datetime DEFAULT NULL,
  PRIMARY KEY (`tx_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8; 

 服务提供者工程导入quartz依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-quartz</artifactId>
        </dependency>

服务提供者yml:

rabbitmq: #rabbitmq配置
  host: 192.168.40.146
  port: 5672
  username: admin
  password: 1111
  virtual-host: /
  publisher-returns: true #开启消息退回回调
  publisher-confirm-type: correlated #开启消息确认回调
  listener:
    direct:
      acknowledge-mode: manual #开启交换机模式手动ack
    simple:
      acknowledge-mode: manual #开启直连模式手动ack

 service:

  1.     @Override
  2.     @Transactional(rollbackFor = {Exception.class})
  3.     public void insertTbItem(TbItemVo tbItem) {
  4.         //修改需要创建修改时间，并且同步修改在关联表tb_item_param_item中
  5.         TbItem param = new TbItem();
  6.         //将工具实体类的值赋值给item对象
  7.         BeanUtils.copyProperties(tbItem,param);
  8.         //设置创建时间
  9.         param.setCreated(new Date());
  10.         param.setUpdated(new Date());
  11.         //设置当前状态
  12.         param.setStatus((byte) 1);
  13.         //设置id
  14.         long id = IDUtils.genItemId();
  15.         param.setId(id);
  16.         //调用mapper修改数据
  17.         tbItemMapper.insert(param);
  18.         //同时修改关联数据
  19.         TbItemParamItem paramItem = new TbItemParamItem();
  20.         //设置参数
  21.         paramItem.setParamData(tbItem.getItemParams());
  22.         //设置关联id
  23.         paramItem.setItemId(id);
  24.         //设置时间
  25.         paramItem.setCreated(new Date());
  26.         paramItem.setUpdated(new Date());
  27.         tbItemParamItemMapper.insert(paramItem);
  28.         //设置关联item_desc
  29.         TbItemDesc desc = new TbItemDesc();
  30.         desc.setItemDesc(tbItem.getDesc());
  31.         desc.setItemId(id);
  32.         desc.setCreated(new Date());
  33.         desc.setUpdated(new Date());
  34.         tbItemDescMapper.insert(desc);
  35.         //保存新增信息至本地消息表
  36.         LocalMessage localMessage = new LocalMessage();
  37.         localMessage.setTxNo(UuidUtils.generateUuid().toString());
  38.         localMessage.setItemId(id);
  39.         localMessage.setState(0);
  40.         localMessageMapper.insertSelective(localMessage);
  41.     }

 这里service执行玩新增后，调用持久层在本地信息记录表中记录新增数据；

编写消息发送类：

  1.  
  2. /**
  3.  * 消息发送者
  4.  */
  5. @Component
  6. public class RabbitMQSender implements ConfirmCallback, ReturnCallback {
  7.  
  8.     private final Logger logger = LoggerFactory.getLogger(RabbitMQSender.class);
  9.  
  10.     @Autowired
  11.     private LocalMessageMapper localMessageMapper;
  12.  
  13.     @Autowired
  14.     private AmqpTemplate amqpTemplate;
  15.  
  16.     private String exchange = String.valueOf(RabbitMQEnum.ITEM_EXCHANGE);
  17.  
  18.     private String routingKey = String.valueOf(RoutingKey.ITEM_INSERT);
  19.  
  20.     /**
  21.      * 发送信息至交换机
  22.      * @param localMessage
  23.      */
  24.     public void sendMsg(LocalMessage localMessage){
  25.         RabbitTemplate rabbitTemplate = (RabbitTemplate) amqpTemplate;
  26.         //设置确认回调和失败回调
  27.         rabbitTemplate.setConfirmCallback(this);
  28.         rabbitTemplate.setReturnCallback(this);
  29.         //创建相关消息对象
  30.         CorrelationData correlationData = new CorrelationData(localMessage.getTxNo());
  31.         //发送信息，需要指定交换机，路由id,发送的信息，相关信息
  32.         rabbitTemplate.convertAndSend(exchange,routingKey, JsonUtils.objectToJson(localMessage));
  33.     }
  34.  
  35.     /**
  36.      * 确认回调方法
  37.      * @param correlationData
  38.      * @param ack
  39.      * @param cause
  40.      */
  41.     @Override
  42.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
  43.         if(ack){
  44.             //消息发送成功，更新本地信息为已发送成功状态，或者直接删除本地记录
  45.             String txNo = correlationData.getId();
  46.             localMessageMapper.deleteByPrimaryKey(txNo);
  47.         }
  48.     }
  49.  
  50.     /**
  51.      * 失败回调方法
  52.      * @param message
  53.      * @param replyCode
  54.      * @param replyText
  55.      * @param exchange
  56.      * @param routingKey
  57.      */
  58.     @Override
  59.     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
  60.             logger.info("return--message:" + new String(message.getBody())
  61.                     + ",exchange:" + exchange + ",routingKey:" + routingKey);
  62.     }
  63. }

如果消息发送成功，则会回调确认回调方法，删除本地信息记录表的数据

失败则调用失败回调方法，打印日志。

编写定时任务类：

  1. /**
  2.  * 任务类，定时发送信息
  3.  */
  4. @Component
  5. public class ItemQuartz {
  6.  
  7.     private Logger logger = LoggerFactory.getLogger(ItemQuartz.class);
  8.  
  9.     @Autowired
  10.     private LocalMessageMapper localMessageMapper;
  11.  
  12.     @Autowired
  13.     private RabbitMQSender rabbitMQSender;
  14.  
  15.     /**
  16.      * 定时任务
  17.      */
  18.     public void scanLocalMessage(){
  19.         logger.info("执行扫描本地消息表的任务：" + new Date());
  20.         List<LocalMessage> localMessages = localMessageMapper.selectByExample(null);
  21.         if(!CollectionUtils.isEmpty(localMessages)){
  22.             for (LocalMessage localMessage : localMessages) {
  23.                 rabbitMQSender.sendMsg(localMessage);
  24.             }
  25.         }
  26.  
  27.     }
  28. }

 定时任务执行扫描操作，如果本地信息表中存在消息，则将其发送。

定时任务配置：

  1. /**
  2.  * 定时任务配置
  3.  */
  4. @Configuration
  5. public class QuartzConfig {
  6.     //定义工作任务
  7.     @Bean
  8.     public MethodInvokingJobDetailFactoryBean methodInvokingJobDetailFactoryBean(ItemQuartz itemQuartz){
  9.         MethodInvokingJobDetailFactoryBean jobDetailFactoryBean =
  10.                 new MethodInvokingJobDetailFactoryBean();
  11.         jobDetailFactoryBean.setTargetObject(itemQuartz);
  12.         jobDetailFactoryBean.setTargetMethod("scanLocalMessage");
  13.         return jobDetailFactoryBean;
  14.     }
  15.     
  16.     //定义触发器
  17.     @Bean
  18.     public CronTriggerFactoryBean cronTriggerFactoryBean(MethodInvokingJobDetailFactoryBean jobDetailFactoryBean){
  19.         CronTriggerFactoryBean triggerFactoryBean = new CronTriggerFactoryBean();
  20.         triggerFactoryBean.setCronExpression("*/1 * * * * ?");
  21.         triggerFactoryBean.setJobDetail(jobDetailFactoryBean.getObject());
  22.         return triggerFactoryBean;
  23.     }
  24.  
  25.     //scheduled：什么时候做什么事
  26.     @Bean
  27.     public SchedulerFactoryBean schedulerFactoryBean(
  28.             CronTriggerFactoryBean triggerFactoryBean) {
  29.         SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
  30.         schedulerFactoryBean.setTriggers(triggerFactoryBean.getObject());
  31.         return schedulerFactoryBean;
  32.     }
  33. }

配置定时任务的执行。

如此便解决了 本地事务与消息发送的原子性问题，以及消息数据持久化的问题（convertAndSent方法发送消息默认持久化）

服务消费者yml配置

spring:
    listener:
      direct:
        acknowledge-mode: manual #手动确认
      simple:
        acknowledge-mode: manual #手动确认

服务消费者（从消息队列中获取消息）

  1. /**
  2.  * 监听消息队列
  3.  */
  4. @Component
  5. public class SearchItemListener {
  6.  
  7.     private Logger logger = LoggerFactory.getLogger(SearchItemListener.class);
  8.  
  9.     @Autowired
  10.     private SearchItemService searchItemService;
  11.  
  12.     @Autowired
  13.     private MsgDistinctService msgDistinctService;
  14.  
  15.     /**
  16.      * 监听mq队列中的消息
  17.      * @param msg 接收到的消息
  18.      * @param channel 信道
  19.      * @param message 封装的信息对象
  20.      */
  21.     @RabbitListener(bindings = {
  22.             @QueueBinding(
  23.                     value = @Queue(value = "item_queue",durable = "true"),
  24.                     exchange = @Exchange(value = "item_exchange",type = ExchangeTypes.TOPIC),
  25.                     key = {"item.*"}
  26.             )
  27.     })
  28.     public void listen(String  msg, Channel channel, Message message) throws IOException {
  29.         logger.info("接收到消息" + msg);
  30.         LocalMessage localMessage = JsonUtils.jsonToPojo(msg, LocalMessage.class);
  31.         //进行幂等判断
  32.         MsgDistinct msgDistinct =msgDistinctService.selectMsgDistinctByTxNo(localMessage.getTxNo());
  33.         if(msgDistinct == null){
  34.             searchItemService.addDoc(localMessage.getItemId());
  35.             msgDistinctService.insertMsgDistinct(localMessage.getTxNo());
  36.         }else{
  37.             System.out.println("=======幂等生效：事务"+msgDistinct.getTxNo()
  38.                     +" 已成功执行===========");
  39.         }
  40.         channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  41.     }
  42. }

 通过注解的方式配置交换机（默认持久）与队列持久化，通过消息去重表保证幂等性，通过手动ack保证消息投递失败时的重新投递，当表里存在本地消息记录表的数据时直接提交，不存在则添加记录并提交。如此便保证了可靠消息最终一致性

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

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

  

闽ICP备14008679号