RocketMQ解决消息发送与业务落库的一致性

2021-04-21 18:07:00 745

在实际业务中, 经常有如下场景

更新一条数据后, 需要发送一条消息异步通知其他服务进行后续处理

比如用户完成激活, 发送一条消息通知第三方服务同步用户状态

那么会有如下代码

@Transactional
public void enabled() {
    //更新用户状态
    updateUser();
    //发送消息
    sendMsg();
}

这样写可能出现这样的问题

事务提交失败, 消息发送成功

那换种方式写

将sendMsg();放在事务外面, 等事务提交成功之后才发送消息

这样也会有问题, 事务提交成功, 但消息发送失败

这两种情况都会导致业务数据出现异常

使用RocketMQ的事务消息就可以解决这个问题.

我们在数据库事务中先发送一个半消息, 这个半消息此时是不能被消费的. 待RocketMQ返回Ack后，再执行本地事务
然后数据库事务结束之后, 再确认数据库事务状态, 确认执行成功了再向RocketMQ发送确认消息, 此时的消息才能算完全体, 可以被Consumer进行消费.
Broker会定时扫描长时间没有进行二次确认的消息, 主动向Producer进行消息回查. 以确定消息是否能正常消费

实现最终一致性

整体机制有点类似MySQL/InnoDB undolog/redolog的二阶段提交

redolog先prepare(innodb_flush_log_at_trx_commit设置成1的情况下, 立即刷盘一次) 该阶段若crash则事务回滚
binlog再写(sync_binlog这个参数设置成1的时候, 每提交1次事务, 都立即持久化到磁盘)立即刷盘该阶段若crash则同样事务回滚
mysql server调用引擎的提交事务接口, redolog 改为commit状态. 该阶段若crash, mysql恢复后会判断binlog中是否已有记录, 若有, 则最终提交事务, 若没有, 则回滚事务

RabbitMQ/RocketMQ消息可靠性保证

RabbitMQRabbitMQ刷盘机制异步写入文件前会有一个Buffer，大小为1M（1048576），数据在写入文件时，首先会写入到这个Buffer，如果Buffer已满，则会将Buffer写入到文件（此时写入到内核态缓存中, 未必刷到磁盘）；有个固定的刷盘时间：25ms，也就是不管Buffer

2022-11-04

自定义rocketmq-spring-boot-starter

一般使用rocketmq-starter都会进行不同程度的封装, 在此分享一种封装思路基于rocketmq-spring-boot-starter <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rock

2022-09-26

RocketMQ解决消息发送与业务落库的一致性

在实际业务中, 经常有如下场景更新一条数据后, 需要发送一条消息异步通知其他服务进行后续处理比如用户完成激活, 发送一条消息通知第三方服务同步用户状态那么会有如下代码@Transactional public void enabled() { //更新用户状态 updateUser(

2021-04-21

RocketMQ架构及特性

架构RocketMQ包含四个组件NameServer, Broker, Consumer, ProducerNameServer类似注册中心, Broker接收存储消息, Consumer和Producer在项目内定义Broker向所有NameServer注册自己, 持续发送心跳包Consumer和

2024-10-23

freemarker 时间显示不正常设置时区

项目在本地开发的时候显示正常，部署上服务器就一直差8个小时，最后发现freemarker官方文档有这样的说明time_zone：时区的名称来显示并格式化时间。默认情况下，使用JVM的时区。也可以是 Java 时区 API 接受的值，或者 "JVM default" (从 FreeMarker 2

2020-03-28

IDEA 2019.1 xml 不高亮

前几天更新了idea后，发现xml里的代码都没有了高亮，变得跟记事本一个德性了打开setting ，搜索 File Types，找到xml项，查看下方的匹配格式，果然没有xml，(idea真是厉害)点击右方的+，输入*.xml，点击ok，解决问题

2020-03-28

npm install 淘宝镜像

npm install --registry=https://registry.npm.taobao.org

2020-03-28

Java中方法的参数传递机制

来看一段代码 public class Man { private String name; private Integer age; public String getName() { return name; } publi

2020-03-28

基于自定义注解手写权限控制

方法一: AOP 方法二: 拦截器项目结构项目依赖<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-w

2020-03-28

Docker 部署详细全过程附代码

Docker 部署本站全过程环境：CentOS7.61. 安装Docker其他版本CentOS可以参考这个https://help.aliyun.com/document_detail/187598.html查看本机内核版本，内核版本需高于 3.10uname -r 确保 yum 包最新yum u

2020-03-28

SpringBoot 启动普通java工程

引入依赖<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> <version>2.0.9</version> </dependency>

2020-03-28

Vue.js DOM操作

<template> <input type="button" @click="reply($event)" value="回复"> </template> export default { methods: { replyFun(e) {

2020-03-29

CentOS7编译调试OpenJDK12

1. 下载源码https://hg.openjdk.java.net/jdk/jdk12点击左侧的browse，再点击zip，就可以下载zip格式的源码压缩包。unzip xxx.zip 解压文件2. 安装jdkyum install java-11-openjdk-devel -y3. 运行con

2020-04-23

编写自己的Spring Boot Starter

1.新建一个maven项目命名规则统一是xxx-spring-boot-starter完整pom.xml<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"

2020-06-29