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

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

背景

为什么要做事务消息中心

原有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依赖

    
        
     
      com.netease.kaola
        
     
      transfer-center-api
        
     
      0.0.2
     
    

实现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
    
      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 consumer

业务中投递消息

    String msgId = kafkaMesageApi.prepare(message);    try{        xxxService.doSomeThing();        message.setMsgId(msgId);    }catch(Exception e){        kafkaMesageApi.rollback(message);     }kafkaMesageApi.commit(message);

注意事项

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

消息体解释

公共参数：

• 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+款云产品！ 

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

相关文章：

【推荐】 

【推荐】