【消息队列】消息中间件RocketMQ4.X急速入门3-阿里云开发者社区

6.消息队列Offset和CommitLog

6.1.RocketMQ消息偏移量Offset

（1）什么是offset

message queue是无限长的数组，一条消息进来下标就会增长1，小标就是offset，消息在某个MessageQueue里的位置，通过offset的值可以定位这条消息，或者指示Consumer从这条消息开始消费。
message queue中maxOffset表示消息的最大offset，maxOffset并不是最新的那条Offset而是新的消息的offset+1，minOffset则是现存的最小的offset
fileReserveTime=48默认消息存储48消息后，消息就会被从磁盘中删除，message queue的min offset也就对应的增长，锁以比minOffset还要小的那些消息已经不在broker上了，就无法消费

（2）类型（父类是OffsetStore）

本地文件类型

DefaultMQPushConsumer的BROADCASTING模式，各个Conusmer没有相互干扰，使用LocalFileOffsetStore，把Offset存储在本地

Broker代存储类型

DefalutMQPushConsumer的CLUSTERING模式，由Broker端存储和控制Offset的值，使用RemoteBrokerOffsetStore

主要是记录消息消费的偏移量，由多个消费者进行消费

集群模式下采用RemoteBrokerOffsetStore，Broker控制offset的值

广播模式下采用LocalFileOffsetStore，消费端存储，消费者控制

建议采用pushConsumer，RocketMQ自动维护OffsetStore，如果是PullConsumer则需要自己维护OffsetStore

6.2.RocketMQ消息存储CommitLog

（1）消息存储是由ConsumeQueue和CommitLog

ConsumeQueue是逻辑队列，CommitLog是真正存储消息文件的，存储的是指向物理存储地址

Topic下的每个message queue都有对应的ConsumeQueue文件，内容也会被持续化到磁盘

默认地址：store/consumequeue/{topicname}/{queueid}/fileName

CommitLog是消息文件的存储地址。CommitLog的生成规则是每个文件默认达到1G开始切割，CommitLog文件名fileName，名字长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1 073 741 824Byte，当着文件满了，第二个文件名字为00000000001073741824，起始便宜量为1073741825=上一个最大的偏移量+1，消息存储的时候会顺序写入，当文件满了则写入下一个文件。

Broker里面多个Topic，一个Topic中有多个MessageQueue，每个MessageQueue对应一个ConusmeQueue，ConsumeQueue里面记录的是消息在CommitLog里面的物理存储地址

6.3.ZeroCopy零拷贝技术

（1）高效原因

CommitLog顺序写，存储了MessageBody、MessageKey、Tag等消息
ConsumeQueue随机读+操作系统PageCache+零拷贝技术ZeroCopy

（2）Linux将一个File文件发送出去（Linux有两个上下文，内核态，用户态）

File经历了4次copy

调用read，将文件从磁盘拷贝到了kernel内核态

CPU控制kernel态的数据copy到用户态

调用write时，user态下的内容会copy到内核态的socket的buffer中

最后将内核态socket buffer的数据copy到网卡设备中传送

（3）ZeroCopy

请求kernel直接把disk的data传输给socket，而不是通过应用程序传输。ZeroCopy大大提高了应用程序的性能，较少不必要的内核缓冲区跟用户缓冲区的拷贝，从而减少CPU的开销和小勺kernel和user模式的上下文切换

（4）对应的零拷贝技术有mmap和sendfile

mmap：小文件传输快

sendfile：大文件传输快

7.RocketMQ实现分布式事务

7.1.分布式事务介绍

（1）什么是分布式事务

来源：单体应用->拆分成分布式应用
一个接口需要调用多个服务，且操作不同的数据库，数据一致性难保证
（2）常见的解决方案

2PC：两阶段提交，基于XA协议
TCC：Try（尝试）、Confirm（证实）、Cancel（取消）

（3）框架

GTS->开源Fescar：地址：https://github.com/alibaba/fescar
LCN：地址：https://github.com/codingapi/tx-lcn

（4）RocketMQ分布式事务消息

RocketMQ提供分布式事务功能，通过RocketMQ事务消息道道分布式事务的一致性。

（5）半消息Half Message

暂时不能消费的消息，是Producer投递到Broker的消息，但是服务端还未收到生产者的二次确认，此时该消息被标记为“暂不能投递状态”，也就是不能投递给消费者，处于该状态的消息为半消息。

（6）整体交互流程

Producer向Broker发送消息

服务端将消息持久化成功后，向发送方ACK确认消息已经发送成功，此时消息为半消息

发送方执行本地事务逻辑

发送方（SpringBoot应用）根据本地事务的结果向服务器发送二次确认（Commit或者Rollback），服务端收到Commit状态则将版消息标记为可投递，订阅放最终消费该消息，如果收到Rollback状态，则订阅方不会消费该消息，三天后Broker进行删除该消息

在断网或者应用重启的情况下，也就是发送方没有给服务器提交二次确认的结果，这回服务端会经过固定时间对发送方的这条消息进行回查

发送方收到消息回查后，需要检查对应用消息的本地事务执行的最终结果

发送方检查得到本地事务的最终状态在进行二次确认，服务端仍按照Commit或者rollback结果进行相应的处理

（7）RocketMQ事务消息的状态

COMMIT_MESSAGE：提交事务的消息，消费者可以消费此消息

ROLLBACK_MESSAGE：回滚事务消息，消息会在broker中删除，消费者不能在消费

UNKNOW：broker需要回查确认消息的状态

（8）关于事务消息的消费

事务消息consumer端的消费方式和普通消息是一样的，RocketMQ能保证消息被consumer收到

7.2.RocketMQ分布式事务消息实战

（1）编写TransactionListenerImpl类实现TransactionListener接口

public class TransactionListenerImpl implements TransactionListener {
    /**
     * 本地事务方法
     * @param msg 消息体
     * @param arg 附加参数
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        System.out.println("executeLocalTransaction.....");
        System.out.println("tag:"+msg.getTags());
        System.out.println("transactionId:"+msg.getTransactionId());
        System.out.println("body:"+msg.getBody().toString());
        if("1".equalsIgnoreCase(arg.toString())){
            return LocalTransactionState.COMMIT_MESSAGE;
        }else if("2".equalsIgnoreCase(arg.toString())){
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }else if("3".equalsIgnoreCase(arg.toString())){
            return LocalTransactionState.UNKNOW;
        }
        return null;
    }
    /**
     * 回查方法
     * @param msg
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("checkLocalTransaction.....");
        System.out.println("tag:"+msg.getTags());
        System.out.println("transactionId:"+msg.getTransactionId());
        System.out.println("body:"+msg.getBody());
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

（2）编写TransactionProdcuer类

@Component
public class TransactionProducer {
    private String producerGroup = "transaction_producer_group";
    private TransactionListener transactionListener = new TransactionListenerImpl();
    private TransactionMQProducer producer;
    /**
     * 设置自定义线程池
     * corePoolSize：池中锁保存的核心线程数
     * maximumPoolSize：池中允许的最大线程数
     * keepActiveTime：非核心线程空闲等待新任务的最长时间
     * timeUnit：keepActiveTime参数的时间单位
     * blockingQueue：任务队列
     */
    ExecutorService executorService = new ThreadPoolExecutor(2, 5,
            100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000),
            new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        }
    });
    public TransactionProducer(){
        producer = new TransactionMQProducer(producerGroup);
        //设置生产者发送broker失败重复发送的次数
        producer.setRetryTimesWhenSendFailed(5);
        //指定NameServer地址，多个地址；隔开
        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        //设置事务监听器
        producer.setTransactionListener(transactionListener);
        //设置线程池
        producer.setExecutorService(executorService);
        start();
    }
    public TransactionMQProducer getProducer(){
        return this.producer;
    }
    /**
     * 对象在使用之前必须调用一次，只能初始化一次
     */
    public void start(){
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown(){
        this.producer.shutdown();
    }
}

（3）TransactionController编写

@RestController
@RequestMapping("api/v1/tran")
public class TransactionController {
    @Autowired
    private TransactionProducer transactionProducer;
    @RequestMapping("/pay_cb1")
    public Object callback1(String tag,String otherArgs) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(JmsConfig.TOPIC,tag,"",tag.getBytes());
        SendResult sendResult = transactionProducer.getProducer().sendMessageInTransaction(message,otherArgs);
        System.out.println(sendResult);
        return new HashMap<>();
    }
}

8.RocketMQ搭建双主双从架构

4台机器，2态部署NameServer，4台部署Broker，双主双从同步双鞋，异步刷盘
环境准备：jdk、maven、rocketmq

（1）机器列表

server1 ssh root@192.168.10.200   部署nameServer  Broker-a
server2 ssh root@192.168.10.201   部署nameServer   Broker-a-s
server3 ssh root@192.168.10.202         Broker-b
server4 ssh root@192.168.10.203         Broker-b-s

（2）修改RocketMQ配置文件（启动内存配置，runbroker.sh和runserver.sh）

vim runserver.sh
JAVA_OPT="${JAVA_OPT} -server -Xms528m -Xmx528m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
vim runbroker.sh
JAVA_OPT="${JAVA_OPT} -server -Xms528m -Xmx528m -Xmn256m"

（3）修改RocketMQ的配置文件并启动

broker-a主节点
nohup sh bin/mqbroker -c conf/2m-2s-sync/broker-a.properties &
namesrvAddr=192.168.159.133:9876;192.168.159.130:9876
brokerClusterName=XdclassCluster
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
defaultTopicQueueNums=4
#是否允许自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=false
#存储路径，根据需求进行配置绝对路径，默认是家目录下面
#storePathRootDir= 
#storePathCommitLog

broker-a从节点
nohup sh bin/mqbroker -c conf/2m-2s-sync/broker-a-s.properties &
namesrvAddr=192.168.159.133:9876;192.168.159.130:9876
brokerClusterName=XdclassCluster
brokerName=broker-a
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
defaultTopicQueueNums=4
#是否允许自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=false
#存储路径，根据需求进行配置绝对路径，默认是家目录下面
#storePathRootDir= 
#storePathCommitLog

broker-b主节点
nohup sh bin/mqbroker -c conf/2m-2s-sync/broker-b.properties &
namesrvAddr=192.168.159.133:9876;192.168.159.130:9876
brokerClusterName=XdclassCluster
brokerName=broker-b
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
defaultTopicQueueNums=4
#是否允许自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=false
#存储路径，根据需求进行配置绝对路径，默认是家目录下面
#storePathRootDir= 
#storePathCommitLog

broker-b从节点
nohup sh bin/mqbroker -c conf/2m-2s-sync/broker-b-s.properties &
namesrvAddr=192.168.159.133:9876;192.168.159.130:9876
brokerClusterName=XdclassCluster
brokerName=broker-b
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
defaultTopicQueueNums=4
#是否允许自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=false
#存储路径，根据需求进行配置绝对路径，默认是家目录下面
#storePathRootDir= 
#storePathCommitLog

（4）双主双从控制台搭建

修改事项
pom.xml  里面的rocketmq版本号
路径：/usr/local/software/rocketmq-externals-master/rocketmq-console/src/main/resources
application.properties  里面的nameServer
增加
rocket.config.namesrvAddr=192.168.159.133:9876;192.168.159.130:9876
mvn install -Dmaven.test.skip=true
java -jar rocketmq-console-ng-1.0.0.jar

（5）RocketMQ生产环境配置

Topic创建上线禁止自动创建
一般是由专门的后台管理队列的CRUD，应用上线需要申请队列名称
生产环境推荐配置
NameServer配置多个不同机器多个节点
多Master每个Master带有Slave
主从设置SYNC_MASTER同步双写
Producer用同部方式投递Broker

刷盘策略为ASYNC_FLUSH

性能思路分析

CPU：top
网卡：sar -n DEV 2 10 、netstat -t

磁盘：iostat -xdm 1
JVM：jstack、MAT、jinfo
闭
autoCreateTopicEnable=true
#是否允许自动创建订阅组，建议线下开启，线上关闭

autoCreateSubscriptionGroup=false

#存储路径，根据需求进行配置绝对路径，默认是家目录下面

#storePathRootDir=

#storePathCommitLog

```bash
broker-b从节点
nohup sh bin/mqbroker -c conf/2m-2s-sync/broker-b-s.properties &
namesrvAddr=192.168.159.133:9876;192.168.159.130:9876
brokerClusterName=XdclassCluster
brokerName=broker-b
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
defaultTopicQueueNums=4
#是否允许自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=false
#存储路径，根据需求进行配置绝对路径，默认是家目录下面
#storePathRootDir= 
#storePathCommitLog

（4）双主双从控制台搭建

修改事项
pom.xml  里面的rocketmq版本号
路径：/usr/local/software/rocketmq-externals-master/rocketmq-console/src/main/resources
application.properties  里面的nameServer
增加
rocket.config.namesrvAddr=192.168.159.133:9876;192.168.159.130:9876
mvn install -Dmaven.test.skip=true
java -jar rocketmq-console-ng-1.0.0.jar

（5）RocketMQ生产环境配置

Topic创建上线禁止自动创建
一般是由专门的后台管理队列的CRUD，应用上线需要申请队列名称
生产环境推荐配置
NameServer配置多个不同机器多个节点
多Master每个Master带有Slave
主从设置SYNC_MASTER同步双写
Producer用同部方式投递Broker
刷盘策略为ASYNC_FLUSH
性能思路分析

CPU：top
网卡：sar -n DEV 2 10 、netstat -t
磁盘：iostat -xdm 1
JVM：jstack、MAT、jinfo

【消息队列】消息中间件RocketMQ4.X急速入门3

6.消息队列Offset和CommitLog

6.1.RocketMQ消息偏移量Offset

6.2.RocketMQ消息存储CommitLog

6.3.ZeroCopy零拷贝技术

7.RocketMQ实现分布式事务

7.1.分布式事务介绍

7.2.RocketMQ分布式事务消息实战

8.RocketMQ搭建双主双从架构

热门文章

最新文章

相关课程

相关电子书

相关实验场景