事务消息中心-TMC

叁叁肆2018-10-29 09:40


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

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


背景

为什么要做事务消息中心

原有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,容器深度实践