`UniqueLongMsgId` 除了 id 以外，还包含其它信息如：从消息从哪个通道账号Id提交的，从哪个IP端口提交的、长短信的分片ID、总分片数、分片序号以及消息序列号、时间戳。
在Test包里有一个模拟的匹配状态报告的测试用例用是用 `UniqueLongMsgId` 实现的，并且经过相同手机号、端口号在极限并发压力下的匹配测试，单JVM多线程安全。逻辑供参考： [`com.zx.sms.transgate.TestReportForward`](./src/test/java/com/zx/sms/transgate/TestReportForward.java)
- `集群环境如何平均分配上游连接数?`
网关平台通常会有多个服务节点，而对接的通道给的连接数通常不是服务节点数的整倍数，极端情况连接数小于服务节点数，这样如何平均分配连接数就成了一个问题。
这里介绍一个算法：通过在redis里记录 {全局的服务节点列表}，来计算每个服务节点连几个tcp连接。
```
var curNodeIndex = getCurNodeIndexFromRedis(thisNode); // 当前节点在全局服务节点的排序号，{0,1,2,3,...}
var cntNode ; // 从Redis里获取的总的服务进程节点数
//所有短信通道，逐一计算每一个通道，在当前节点上最大允许的tcp连接数，
allEntityPointList.foreach(e->{
var curEntityIndex = getCurEntityIndex(e); //所有短信通道根据Id排序后，当前通道的排序号，{0,1,2,3,4,5,6,7,...}
var curMaxChannel = e.getMaxChannel(); //当前通道全局允许的最大连接数
//连接数不是服务节点数的整倍数，按服务节点数平均分配后一定会有余数， 按当前节点的排序号先后把余下的连接数分完。
//但是服务节点排序号是固定不变的，这样排序号靠前的节点总是优先分到余下的连接数，造成全局通道总连接数分配不均，因此要结合"当前通道的排序号" 对 "服务节点排序号"进行位移
//因此，当"服务节点"或者"全局通道账号"有任一个变化时，都会影响连接的分配。
var shiftNodeIndex = (curNodeIndex + curEntityIndex) % cntNode;
//平均分配后,余下的连接数
var remainderChannel = curMaxChannel % cntNode;
//平均分配连接数
var hostMaxChannel = curMaxChannel / cntNode;
//余数处理
if(remainderChannel > 0 && shiftNodeIndex < remainderChannel){
hostMaxChannel = hostMaxChannel + 1;
}
var hostChannel = getConnectionCountAtCurrentNode(e); //当前通道在本节点上的连接数
if(hostChannel < hostMaxChannel){
openChannel(e); //新建一个连接
}else{
//关闭该通道超过数量的连接
closeSomeChannel(e,hostMaxChannel - hostChannel);
}
});
```
- `框架内部的netty的Handler前后顺序`
如图：
# CMPPGate , SMPPGate , SGIPGate, SMGPGate
中移短信cmpp协议/smpp协议 netty实现编解码
这是一个在netty4框架下实现的cmpp3.0/cmpp2.0短信协议解析及网关端口管理。
代码copy了 `huzorro@gmail.com` 基于netty3.7的cmpp协议解析 [huzorro@gmail.com 的代码 ](https://github.com/huzorro/netty3ext)
目前已支持发送和解析`长文本短信拆分合并`，`WapPush短信`，以及`彩信通知`类型的短信。可以实现对彩信或者wap-push短信的拦截和加工处理。wap短信的解析使用 [smsj](https://github.com/marre/smsj) 的短信库
cmpp协议已经跟华为，东软，亚信的短信网关都做过联调测试，兼容了不同厂家的错误和异常，如果跟网关通信出错，可以打开trace日志查看二进制数据。
因要与短信中心对接，新增了对SMPP协议的支持。
SMPP的协议解析代码是从 [Twitter-SMPP 的代码](https://github.com/fizzed/cloudhopper-smpp) copy过来的。
新增对sgip协议(联通短信协议)的支持
sgip的协议解析代码是从 [huzorro@gmail.com 的代码 ](https://github.com/huzorro/sgipsgw) copy过来后改造的。
新增对smgp协议(电信短信协议)的支持
smgp的协议解析代码是从 [SMS-China 的代码 ](https://github.com/clonalman/SMS-China) copy过来后改造的。
支持发送彩信通知，WAP短信以及闪信(Flash Message):
## 性能测试
在48core，128G内存的物理服务器上测试协议解析效率：35K条/s, cpu使用率25%.
## Build
执行mvn package . jdk1.6以上.
## 增加了业务处理API
业务层实现接口：BusinessHandlerInterface，或者继承AbstractBusinessHandler抽象类实现业务即可。 连接保活，消息重发，消息持久化，连接鉴权都已封装，不须要业务层再实现。
## 如何实现自己的Handler,比如按短短信计费
参考 CMPPChargingDemoTest 里的扩展位置
# 实体类说明
## CMPP的连接端口
`com.zx.sms.connect.manager.cmpp.CMPPEndpointEntity`
表示一个Tcp连接的发起端，或者接收端。用来记录连接的IP.port,以及CMPP协议的用户名，密码，业务处理的ChannelHandler集合等其它端口参数。包含三个子类：
1. com.zx.sms.connect.manager.cmpp.CMPPServerEndpointEntity
服务监听端口，包含一个List属性。 一个服务端口包含多个CMPPServerChildEndpointEntity端口
2. com.zx.sms.connect.manager.cmpp.CMPPServerChildEndpointEntity
服务接收端口，包含CMPP连接用户名，密码，以及协议版本等信息
3. com.zx.sms.connect.manager.cmpp.CMPPClientEndpointEntity
客户端端口，包含CMPP连接用户名，密码，以及协议版本，以及服务端IP.port. 用于连接服务端
## 端口连接器接口
`com.zx.sms.connect.manager.EndpointConnector`
负责一个端口的打开，关闭，查看当前连接数，新增连接，移除连接。每个端口的实体类都对应一个EndpointConnector.当CMPP连接建立完成，将连接加入连接器管理，并给pipeLine上挂载业务处理的ChannelHandler.
1. com.zx.sms.connect.manager.cmpp.CMPPServerEndpointConnector
这个类的open()调用netty的ServerBootstrap.bind()开一个服务监听
2. com.zx.sms.connect.manager.cmpp.CMPPServerChildEndpointConnector
用来收集CMPPServerChildEndpointEntity端口下的所有连接。它的open()方法为空.
3. com.zx.sms.connect.manager.cmpp.CMPPClientEndpointConnector
这个类open()调用netty的Bootstrap.connect()发起一个TCP连接
## 端口管理器
`com.zx.sms.connect.manager.EndpointManager`
该类是单例模式，管理所有端口，并负责所有端口的打开，关闭，以及端口信息保存，以及连接断线重连。
## CMPP协议的连接登陆管理
`com.zx.sms.session.cmpp.SessionLoginManager`
这是一个netty的ChannelHandler实现，主要负责CMPP连接的建立。当CMPP连接登陆成功、会话建立完成后，会调用EndpointConnector.addChannel(channel)方法，把连接加入连接器管理，连接器负责给channel的pipeline上挂载业务处理的Handler,最后触发
SessionState.Connect事件，通知业务处理Handler连接已建立成功。
## CMPP的连接状态管理器
`com.zx.sms.session.cmpp.SessionStateManager`
这是一个netty的ChannelHandler实现。负责每个连接上CMPP消息的存储，短信重发，流量窗口控制，过期短信的处理
## CMPP协议解析器
CMPP20MessageCodecAggregator [2.0协议]
CMPPMessageCodecAggregator [这是3.0协议]
聚合了CMPP主要消息协议的解析，编码，长短信拆分，合并处理。
## 短信持久化存储实现 StoredMapFactory
使用BDB的StoreMap实现消息持久化，防止系统意外丢失短信。
## 程序启动处理流程
1. 程序启动类 new 一个CMPPEndpointEntity的实体类并设置IP,port,用户名，密码，业务处理的Handler等参数,
2. 程序启动类 调用EndpointManager.addEndpointEntity(endpoint)方法，将端口加入管理器
3. 程序启动类 调用EndpointManager.openAll()或者EndpointManager.openEndpoint()方法打开端口。
4. EndpointManager会调用EndpointEntity.buildConnector()创建一个端口连接器，并调用EndpointConnector.open()方法打开端口。
5. 如果是CMPPClientEndpointEntity的话，就会向服务器发起TCP连接请求，如果是CMPPServerEndpointEntity则会在本机开启一个服务端口等客户端连接。
6. TCP连接建立完成后。netty会调用EndpointConnector.initPipeLine()方法初始化PipeLine，把CMPP协议解析器，SessionLoginManager加到PipeLine里去，然后netty触发ChannelActive事件。
7. 在SessionLoginManager类里，客户端收到ChannelActive事件后会发送一个CMPPConnnect消息，请求建立CMPP连接.
8. 同样在SessionLoginManager.channelRead()方法里,服务端会收到CMPPConnnect消息，开始对用户名，密码进行鉴权，并给客户端返回鉴权结果。
9. 鉴权通过后，SessionLoginManager调用EndpointConnector.addChannel(channel)方法，把channel加入ArrayList,并给pipeLine上挂载SessionStateManager和业务处理的ChannelHandler，如心跳处理，日志记录，长短信合并拆分处理类。
10. EndpointConnector.addChannel(channel)完成后，SessionLoginManager调用ctx.fireUserEventTriggered()方法，触发 SessionState.Connect事件。
以上CMPP连接建立完成。
11. 业务处理类收到SessionState.Connect事件，开始业务处理，如从MQ获取短信下发，或开启Consumer接收MQ推送的消息。
12. SessionStateManager会拦截所有read()和write()的消息，进行消息持久化，消息重发，流量控制。
## 增加同步调用api
smsgate自开发以来，一直使用netty的异步发送消息，但实际使用场景中同步发送消息的更方便，或者能方便的取到response。因此增加一个同步调用的api。即：发送消息后等接收到对应的响应后才返回。
使用方法如下：
```java
//因为长短信要拆分，因此返回一个promiseList.每个拆分后的短信对应一个promise
List futures = ChannelUtil.syncWriteLongMsgToEntity("client",submitmessage);
for(Promise future: futures){
//调用sync()方法，阻塞线程。等待接收response
future.sync();
//接收成功，如果失败可以获取失败原因，比如遇到连接突然中断错误等等
if(future.isSuccess()){
//打印收到的response消息
logger.info("response:{}",future.get());
}else{
打印错误原因
logger.error("response:{}",future.cause());
}
}
//或者不阻塞进程，不调用sync()方法。
List promises = ChannelUtil.syncWriteLongMsgToEntity("client",submitmessage);
for(Promise promise: promises){
//接收到response后回调Listener方法
promise.addListener(new GenericFutureListener() {
@Override
public void operationComplete(Future future) throws Exception {
//接收成功，如果失败可以获取失败原因，比如遇到连接突然中断错误等等
if(future.isSuccess()){
//打印收到的response消息
logger.info("response:{}",future.get());
}else{
打印错误原因
logger.error("response:{}",future.cause());
}
}
});
}
```
## CMPP Api使用举例
```java
public class TestCMPPEndPoint {
private static final Logger logger = LoggerFactory.getLogger(TestCMPPEndPoint.class);
@Test
public void testCMPPEndpoint() throws Exception {
ResourceLeakDetector.setLevel(Level.ADVANCED);
final EndpointManager manager = EndpointManager.INS;
CMPPServerEndpointEntity server = new CMPPServerEndpointEntity();
server.setId("server");
server.setHost("127.0.0.1");
server.setPort(7890);
server.setValid(true);
//使用ssl加密数据流
server.setUseSSL(false);
CMPPServerChildEndpointEntity child = new CMPPServerChildEndpointEntity();
child.setId("child"); //自定义通道账号ID，保持全局唯一
child.setChartset(Charset.forName("utf-8"));
child.setGroupName("test"); //自定义通道账号分组ID，用于对通道标识不同组，方便路由实现
child.setUserName("901783"); //通道账号，可能和企业代码相同
child.setPassword("ICP001"); //密码
child.setValid(true);
child.setVersion((short)0x30); //协议版本号，48是3.0 协议，32是2.0协议
child.setMaxChannels((short)4);
child.setRetryWaitTimeSec((short)30);
child.setMaxRetryCnt((short)3);
child.setReSendFailMsg(true);
// child.setWriteLimit(200);
// child.setReadLimit(200);
List serverhandlers = new ArrayList();
serverhandlers.add(new CMPPMessageReceiveHandler()); //在这个handler里接收短信
child.setBusinessHandlerSet(serverhandlers);
server.addchild(child);
manager.addEndpointEntity(server);
CMPPClientEndpointEntity client = new CMPPClientEndpointEntity();
client.setId("client"); //自定义通道账号ID，保持全局唯一
client.setHost("127.0.0.1");
// client.setLocalhost("127.0.0.1");
// client.setLocalport(65521);
client.setPort(7890);
client.setChartset(Charset.forName("utf-8"));
client.setGroupName("test"); //自定义通道账号分组ID，用于对通道标识不同组，方便路由实现
client.setUserName("901783"); //通道账号，可能和企业代码相同
client.setPassword("ICP001"); //密码
client.setMsgSrc("902176"); //企业代码 ，可能和UserName相同
client.setSpCode("10658762"); //服务代码，即显示到手机上的号码
client.setMaxChannels((short)10); //最大连接数
client.setVersion((short)0x30); //协议版本号
client.setRetryWaitTimeSec((short)30);//发送request后 等待N秒后没有收到response，则重发消息
client.setMaxRetryCnt((short)3); // 发送消息的最大次数，如果为3，则表示连带第一次发送，再重试2次，一共发送3次
client.setCloseWhenRetryFailed(false); // 当发送消息次数达到最大（MaxRetryCnt）后 ，是否关闭连接。默认是 true 关闭连接
client.setUseSSL(false); //是否使用SSL加密连接，默认为false，不加密
client.setWriteLimit(100); //发送request消息的最大速度（单位条数）
client.setReadLimit(100); //接收request的最大速度（单位条数），当消息量超过一定限制后，消息将积压在TCP网络协议栈的接收缓冲区
client.setWindow(16); //设置发送消息的滑动窗口，滑动窗口默认为16，该大小根据网络时延不同，会影响发送速度
//默认为false ,发送request是否保存在本地磁盘。如果为true，当进程关闭后，本地磁盘会保存未收到response的消息，当进程再次启动框架自动读取消息并发送。可能造成消息重复发送
client.setReSendFailMsg(true);
client.setSupportLongmsg(SupportLongMessage.BOTH); //是否支持长短信的自动拆分和拼接
List clienthandlers = new ArrayList();
clienthandlers.add( new CMPPSessionConnectedHandler(10000)); //在这个handler里发送短信
client.setBusinessHandlerSet(clienthandlers);
manager.addEndpointEntity(client);
manager.openEndpoint(server);
Thread.sleep(1000);
for(int i=0;i<=child.getMaxChannels()+1;i++)
manager.openEndpoint(client);
System.out.println("start.....");
// Thread.sleep(300000);
LockSupport.park();
EndpointManager.INS.close();
}
}
```
## SMPP Api使用举例
```java
public class TestSMPPEndPoint {
private static final Logger logger = LoggerFactory.getLogger(TestSMPPEndPoint.class);
@Test
public void testSMPPEndpoint() throws Exception {
final EndpointManager manager = EndpointManager.INS;
SMPPServerEndpointEntity server = new SMPPServerEndpointEntity();
server.setId("smppserver");
server.setHost("127.0.0.1");
server.setPort(2776);
server.setValid(true);
//使用ssl加密数据流
server.setUseSSL(false);
SMPPServerChildEndpointEntity child = new SMPPServerChildEndpointEntity();
child.setId("smppchild");
child.setSystemId("901782");
child.setPassword("ICP");
child.setValid(true);
child.setChannelType(ChannelType.DUPLEX);
child.setMaxChannels((short)3);
child.setRetryWaitTimeSec((short)30);
child.setMaxRetryCnt((short)3);
child.setReSendFailMsg(true);
child.setIdleTimeSec((short)15);
// child.setWriteLimit(200);
// child.setReadLimit(200);
List serverhandlers = new ArrayList();
serverhandlers.add(new SMPPSessionConnectedHandler(10000));
child.setBusinessHandlerSet(serverhandlers);
server.addchild(child);
SMPPClientEndpointEntity client = new SMPPClientEndpointEntity();
client.setId("smppclient");
client.setHost("127.0.0.1");
client.setPort(2776);
client.setSystemId("901782");
client.setPassword("ICP");
client.setChannelType(ChannelType.DUPLEX);
client.setMaxChannels((short)12);
client.setRetryWaitTimeSec((short)100);
client.setUseSSL(false);
client.setReSendFailMsg(true);
// client.setWriteLimit(200);
// client.setReadLimit(200);
client.setSupportLongmsg(SupportLongMessage.SEND); //接收长短信时不自动合并
List clienthandlers = new ArrayList();
clienthandlers.add( new SMPPMessageReceiveHandler());
client.setBusinessHandlerSet(clienthandlers);
manager.addEndpointEntity(server);
manager.addEndpointEntity(client);
manager.openAll();
manager.startConnectionCheckTask();
Thread.sleep(1000);
for(int i=0;i serverhandlers = new ArrayList();
serverhandlers.add(new SgipReportRequestMessageHandler());
serverhandlers.add(new SGIPMessageReceiveHandler());
child.setBusinessHandlerSet(serverhandlers);
server.addchild(child);
manager.addEndpointEntity(server);
SgipClientEndpointEntity client = new SgipClientEndpointEntity();
client.setId("sgipclient");
client.setHost("127.0.0.1");
client.setPort(8001);
client.setLoginName("333");
client.setLoginPassowrd("0555");
client.setChannelType(ChannelType.DUPLEX);
client.setMaxChannels((short)10);
client.setRetryWaitTimeSec((short)100);
client.setUseSSL(false);
client.setReSendFailMsg(true);
// client.setWriteLimit(200);
// client.setReadLimit(200);
List clienthandlers = new ArrayList();
clienthandlers.add(new SGIPSessionConnectedHandler(10000));
client.setBusinessHandlerSet(clienthandlers);
manager.addEndpointEntity(client);
manager.openAll();
Thread.sleep(1000);
for(int i=0;i