事务消息中心-TMC

此文已由作者杨凯明授权网易云社区发布。

欢迎访问网易云社区，了解更多网易技术产品运营经验。

背景

为什么要做事务消息中心

原有kqueue的方式缺点:

• 降低业务库性能

• 占用业务库磁盘

• 历史数据管理成本高

• 技术运维工作量很大

事务消息中心的优点：

• 去除上述kqueue的缺点

• 易运维

• 易扩展

事务消息中心的缺点:

• 接入适当改造

• 增加网络开销

TMC基本概念

事务消息中心主要是提供两阶段提交的方案，对业务方消息提供保证投递的支持。

• 支持RabbitMQ及Kafka

• 支持多数据源、多类型

• 服务端水平扩展

• 不同业务拆库拆表

名词解释

Application：应用，对应CMDB上的应用名。

TableName：表名，一类消息可以存在一个消息表中，一个消息表可以对应多个exchange或者topic，一个TableName会创建一个Retry Thread及一个CallBack Thread。

TaskItem：任务项，代表一个topic或者一个exchange，一个TaskItem会创建6个Send Thread。

Send Thread：发送线程，每个TaskItem会有创建指定数量的发送线程,可以是Kafka的发送线程，也可以说RabbitMQ的发送线程。。

State Thread：状态线程，是处理消息队列comfirm的线程，包括Kafka、RabbitMQ。。

Retry Thread：重试线程，是处理投递失败消息的线程，重试次数默认是3次。如果发送线程内存队列积满，会导致消息只落库，不投递，也需要重试线程来发送。。

CallBack Thread：回查线程，是处理prepare状态的消息，默认取前3分钟，状态还是prepare状态的消息调用客户端回查接口进行确认。

消息中心平台架构图

• 哨兵监控，引入哨兵监控图表展示

• 应用同步自CMDB

• 应用绑定数据源

• 应用按业务划分TableName

• TableName支持一种消息实例多个TaskItem

• 环境按TableName或者TaskItem分离

• 数据源管理

消息中心服务端架构图

• API层，TMC对外暴露Dubbo接口，提供Kafka API以及RabbitMQ API。

• 引擎层，主要分为4个模块

• 发送模块：初始化时会根据配置的TaskItem创建Send Thread和State Thread，默认是各6个Thread，

• 重试模块：初始化时会根据TableName维度创建Retry Thread，每个TableName全局一个重试线程。

• 回查模块：初始化时会根据TableName维度创建CallBack Thread，每个TableName全局一个回查线程，回查接口通过泛化调用客户端DUBBO接口。

• 哨兵模块：哨兵会定时收集内存队列大小、发送线程数、状态线程数、重试线程数、回查线程数等

• 存储层，根据消息系统的类型以及数据库的类型，创建出对应的MessageStore，用以支持Kafka、RabbiMQ以及Mysql、DDB、ORACLE。

• 基础层，可以是Mysql，DDB或者ORACLE，并部署了一个zookeeper来支持重试线程以及回查线程的选主及配置刷新。

业务方-服务端的投递时序图

• 除了prepare接口外，其他接口可异步

接入准备

申请消息中心配置创建（待规范）

1.环境说明

2.appName： 应用名称，同CMDB

3.tableName： 在事务消息中心DB表，表名可以自己定义，也可以由我定义后给你

4.topic： kafka的topic  或者 exchange： RabbitMQ的exchange

• 队列的创建按照原有方式，kafka目前在kafkamanager上建，RabbitMQ走mq申请邮件

• 前期优化及功能更新较为频繁，具体信息可以popo沟通，手把手支持：hzyangkaiming@corp.netease.com

• 区分测试的各个环境 以及pre、beta、online可以通过tableName 或者 topic及exchange来做，具体情况具体分析。

中心配置简图如下：  

客户端接入(java+dubbo)

引入pom依赖

<dependency>
   <groupId>com.netease.kaola</groupId>
   <artifactId>transfer-center-api</artifactId>
   <version>0.0.2</version></dependency>

实现callback接口

a.先继承jar包中的接口

package com.netease.kaola.message.client;import com.netease.kaola.message.api.CallBackFacede;public interface CallBackService extends CallBackFacede{

}

b.实现自己的回查接口实现

package com.netease.kaola.message.client;import java.util.List;import com.netease.kaola.message.api.meta.CallBackResult;@Service("callBackService")public class CallBackServiceImpl implements CallBackService{    @Override
    public CallBackResult callback(List<String> primaryKyes) {
        CallBackResult result = new CallBackResult();        // 这里会搜到主键Keys列表，业务进行一一确认后返回
        for(int i = 0;i < primaryKyes.size() ; i++){            if(primaryKyes.size() > 1 && i == (primaryKyes.size() -1) ){
                result.addRollBack(primaryKyes.get(i));
            }else{
                result.addCommit(primaryKyes.get(i));
            }
        }        return result;
    }

}

配置callback的dubbo provider

<dubbo:service interface="com.netease.kaola.message.client.CallBackService" ref="callBackServiceImpl" />

配置两阶段提交接口dubbo consumer

<dubbo:reference id="kafkaMesageApi" interface="com.netease.kaola.message.api.KafkaMessageApiService" /><dubbo:reference id="rabbitmqMesageApi" interface="com.netease.kaola.message.api.RabbitmqMessageApiService" />

业务中投递消息

    String msgId = kafkaMesageApi.prepare(message);    try{
        xxxService.doSomeThing();
        message.setMsgId(msgId);

    }catch(Exception e){
        kafkaMesageApi.rollback(message); 
    }
kafkaMesageApi.commit(message);

注意事项

• a.对RT较敏感的服务，可以配置commit及rollback接口异步调用。

  <dubbo:reference id="kafkaMesageApi" interface="com.netease.kaola.message.api.KafkaMessageApiService">
   <dubbo:method name="commit" async="true" />
   <dubbo:method name="rollback" async="true" /></dubbo:reference>

消息体解释

公共参数：

• msgId：消息id,服务端生成,commit时需要传

• primaryKey：业务主键字段,由业务方自己填入，回查时的业务主键

• serverName：业务应用的服务器名称

• appName：应用名称，同CMDB

• tableName：在事务消息中心DB表

• payload：消息内容

KFMessageDTO：

• topic：kafka的topic

RBMessageDTO：

• exchange：RabbitMQ的exchange

• routingKey：RabbitMQ的routingKey

哨兵采集器TransferCollector

实现了哨兵采集器，用来监控TMC事务消息中心的各项指标

网易云免费体验馆，0成本体验20+款云产品！ 

更多网易技术、产品、运营经验分享请点击。

相关文章：
【推荐】 Android之高效率截图
【推荐】 IT和非IT人士：2分钟了解什么是区块链
【推荐】 [5.19线下活动]DockerMeetup杭州站—拥抱Kubernetes，容器深度实践