硬件设备
首先是硬件设备，这类短信模块，modem pool大都是基于Q2406A, Q2406B之类的串口设备，只支持GSM和GPRS，不支持电信CDMA，早先的设备只有COM口，如果是一个pool，对应每一个模块都会引出一个COM口，后来出的设备改成了USB2.0接口，其芯片主要是PL2303系列的USB2 Serial Comm方案。在连接到主机后，每一个模块都会显示为一个单独的COM口，COM口编号根据系统当前的计数自动增长。
硬件驱动
Linux
绝大多数发行版都自带PL2303的驱动，插上即可识别，可以通过dmesg看到硬件变化
[ 6096.417435] usb 2-2.1.1: New USB device found, idVendor=067b, idProduct=2303, bcdDevice= 3.00 [ 6096.417437] usb 2-2.1.1: New USB device strings: Mfr=1, Product=2, SerialNumber=0 [ 6096.417439] usb 2-2.1.1: Product: USB-Serial Controller [ 6096.417440] usb 2-2.1.1: Manufacturer: Prolific Technology Inc. [ 6096.424963] pl2303 2-2.1.1:1.0: pl2303 converter detected [ 6096.426240] usb 2-2.1.1: pl2303 converter now attached to ttyUSB0
通过 lsusb, 也可以看到具体的vendor, idproduct等信息
us 002 Device 045: ID 067b:2303 Prolific Technology, Inc. PL2303 Serial Port Bus 002 Device 044: ID 067b:2303 Prolific Technology, Inc. PL2303 Serial Port Bus 002 Device 043: ID 067b:2303 Prolific Technology, Inc. PL2303 Serial Port Bus 002 Device 042: ID 067b:2303 Prolific Technology, Inc. PL2303 Serial Port
配置udev规则
这一步很重要, 否则执行java代码查找设备时会无法看到comm设备, 即执行 CommPortIdentifier.getPortIdentifiers() 代码时返回的结果数为0. 参考的解决方案在这里.
Create a file /etc/udev/rules.d/51-my_usb_device (for instance) 这个文件名可以自己定义, 但是最好以 51- 开头以确保执行顺序. And put the following line:
SUBSYSTEMS=="usb", ATTRS{idVendor}=="067b", ATTRS{idProduct}=="2303", GROUP="users", MODE="0666"
然后关闭COMM设备电源后重新加电, 就能看到变化了, 对应的mod变成了666, 用户组变成了users
$ ll /dev/ttyU* crw-rw-rw- 1 root users 188, 0 Aug 23 15:20 /dev/ttyUSB0 crw-rw-rw- 1 root users 188, 1 Aug 23 15:20 /dev/ttyUSB1 crw-rw-rw- 1 root users 188, 2 Aug 23 15:20 /dev/ttyUSB2 crw-rw-rw- 1 root users 188, 3 Aug 23 15:56 /dev/ttyUSB3
Windows
Win32未测试，主要是在Win64。对于PL2303系列芯片，Windows会自动安装驱动，如果未自动安装的可以手动安装。
遇到的问题：在Windows 10下PL2303设备会显示为“pl2303hxa自2012已停产,请联系供货商”，或者"pl2303hxa phased out since 2012"，无法使用，此时需要安装旧版驱动，并手动选择使用旧驱动。搜索 PL2303_Prolific_DriverInstaller_3.3.3.zip 或者 PL2303_Prolific_GPS_1013_20090319.zip，推荐使用前者。
下载后执行安装程序，待安装结束后，在Device Manager -> Ports 下对应的串口上右键，点击Update driver，点击 Browse my computer for drivers，在下一步点击“Let me pick from a list of available drivers from my computer"，在里面选择日期为2009(对应3.3.3)，或2008(对应3.3.2)的驱动，下一步确认后，Ports列表里的串口设备都会一致变更为Prolific USB-to-Serial Comm Port，并展示COM口编号。
Windows 10重启后, 再次接入硬件时, 依然会使用日期较新的驱动, 需要手动再操作 Update driver.
JAVA配置(方案一)
现在无论是开发环境，还是生产环境，已经全面使用64位操作系统，推荐直接使用RXTX的"rxtx 2.2pre2 (prerelease)"版本，下载地址 里面包含了Windows和Linux 64位版本的jar, dll 和 so. 在jLog的页面上也有可用的dll和jar推荐，经过CRC SHA-256比对，jLog提供的jar和dll和2.2pre2里的文件是一样的。
如果没有将文件放入正确目录，运行时会出现 java.lang.NoClassDefFoundError: gnu/io/CommPortIdentifier 错误。
Linux
Linux下一般只用JDK, 需要在JDK目录下放置so和jar文件, 确认自己的java环境对应的目录, 然后
1. 将RXTXcomm.jar 复制到 [JAVA_HOME]/jre/lib/ext/, 例如我使用的是 /opt/jdk/jdk1.8.0_251/jre/lib/ext/
2. 将librxtxSerial.so 复制到[JAVA_HOME]/jre/lib/amd64/, 例如我使用的是 /opt/jdk/jdk1.8.0_251/jre/lib/amd64/
如果没有正确放置so文件, 运行时会报错 java.lang.UnsatisfiedLinkError: no rxtxSerial in java.library.path
Windows
需要在JDK和JRE目录下都放置dll和jar文件，首先确认自己环境下java对应的安装目录，然后
1. 将RXTXcomm.jar复制到
C:Program FilesJavajre1.8.0_251libext
C:Program FilesJavajdk1.8.0_251jrelibext
2. 将rxtxSerial.dll复制到
C:Program FilesJavajre1.8.0_251in
C:Program FilesJavajdk1.8.0_251jrein
JAVA配置(方案二)
使用 nrjavaserial , 这个jar中包含了各个操作系统的native文件, 不需要在JDK中添加dll, so和jar. 经过测试完全可以替代方案一. 在dependency中加入jar依赖就可以了.
这个方案的另一个优点在于, 提供了arm和arm64 linux 的支持, 并且支持串口设备掉电通知.
JAVA程序
Java程序都是基于smslib v3的库方法. 使用的版本为3.5.4
Serial COMM端口和波特率检测
通过这段代码，可以列出机器上可用的COMM口及其对应的波特率，以及上面的设备信息
public class CommUtil {
private static final Logger logger = LoggerFactory.getLogger(CommUtil.class);
private static final int[] baudRates = {9600, 19200, 57600, 115200};
public static List<CommPortDevice> getCommPortDevices() {
//System.setProperty("gnu.io.rxtx.SerialPorts", "/dev/ttyUSB0");
List<CommPortDevice> devices = new ArrayList<>();
logger.info("Detecting serial comm devices...");
Enumeration<CommPortIdentifier> portList = CommPortIdentifier.getPortIdentifiers();
while (portList.hasMoreElements()) {
CommPortIdentifier portId = portList.nextElement();
if (portId.getPortType() == CommPortIdentifier.PORT_SERIAL) {
logger.info("Found serial comm device: {}", portId.getName());
for (int baudRate : baudRates) {
logger.info(" Trying at {} ...", baudRate);
SerialPort serialPort = null;
try {
serialPort = portId.open("SMSLibCommTester", 1971);
serialPort.setFlowControlMode(SerialPort.FLOWCONTROL_RTSCTS_IN);
serialPort.setSerialPortParams(baudRate, SerialPort.DATABITS_8, SerialPort.STOPBITS_1, SerialPort.PARITY_NONE);
InputStream inStream = serialPort.getInputStream();
OutputStream outStream = serialPort.getOutputStream();
serialPort.enableReceiveTimeout(1000);
int c;
while ((c = inStream.read()) != -1) {
// do nothing;
}
outStream.write("AT
".getBytes());
StringBuilder sb = new StringBuilder();
while ((c = inStream.read()) != -1) {
sb.append((char)c);
}
if (!sb.toString().contains("OK")) {
logger.info(" No device detected, response: {}", sb);
} else {
logger.info(" Device found,");
sb.setLength(0); // clear the StringBuilder
outStream.write("AT+CGMM
".getBytes());
while ((c = inStream.read()) != -1) {
sb.append((char)c);
}
String model = sb.toString()
.replaceAll("(
|
|AT\+CGMM|OK)", "")
.replaceAll("\s+", " ").trim();
logger.info(" Device model: {}", model);
CommPortDevice device = new CommPortDevice();
device.setSerialPort(portId.getName());
device.setBaudRate(baudRate);
device.setModel(model);
devices.add(device);
break;
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
if (serialPort != null) {
serialPort.close();
}
}
}
}
}
return devices;
}
public static void main(String[] args) {
List<CommPortDevice> devices = getCommPortDevices();
logger.info("size: {}", devices.size());
}
}
使用service.sendMessage()方法发送短信
sendMessage()使用的是同步的发送方式, 当服务添加了多个gateway时, sendMessage()使用默认的Round Robin轮询发送, 依次循环使用每一个gateway, 一次发送耗时大约3~5秒. 如果自己管理gateway路由也许可以多个gateway并行发送. 因为现在运营商对短信发送的频率有限制, 上限为200/hour, 1000/day(节假日500/hour, 2000/day), 均分到每分钟的可发短信其实很少, 如果gateway数量不超过4个, 并行意义不大.
这里的代码加了两个列表, 分别用于记录inbound和outbound的历史, 如果需要在重启后依然保持, 可以换成mysql或者redis.
public class ModemService {
private static final Logger logger = LoggerFactory.getLogger(ModemService.class);
private static final int HISTORY_SIZE = 5000;
private final Service service = Service.getInstance();
private final LinkedList<OutboundMessage> outboundMessages = new LinkedList<>();
private final LinkedList<InboundMessage> inboundMessages = new LinkedList<>();
private boolean ready = false;
synchronized public void init() {
service.setOutboundMessageNotification(new OutboundNotification());
service.setInboundMessageNotification(new InboundNotification(inboundMessages));
service.setOrphanedMessageNotification(new OrphanedNotification());
service.setCallNotification(new CallNotification());
service.setGatewayStatusNotification(new GatewayStatusNotification());
List<CommPortDevice> devices = CommUtil.getCommPortDevices();
List<SerialModemGateway> modemGateways = new ArrayList<>();
for (CommPortDevice device : devices) {
logger.info("Add gateway: {},{}", device.getSerialPort(), device.getBaudRate());
SerialModemGateway gateway = new SerialModemGateway(
"modem." + device.getSerialPort(), device.getSerialPort(), device.getBaudRate(), "WAVECOM", "");
gateway.setInbound(true);
gateway.setOutbound(true);
try {
service.addGateway(gateway);
modemGateways.add(gateway);
} catch (GatewayException e) {
logger.error(e.getMessage(), e);
}
}
// 启用轮循模式
service.S.SERIAL_POLLING = true;
// 关闭运营商选择
service.S.DISABLE_COPS = true;
try {
logger.info("ModemService starting");
service.startService();
ready = true;
logger.info("ModemService started");
} catch (SMSLibException |IOException|InterruptedException e) {
logger.error(e.getMessage(), e);
}
for (SerialModemGateway gateway : modemGateways) {
try{
logger.debug("Gateway: {}", gateway.getGatewayId());
logger.debug(" Manufacturer: {}", gateway.getManufacturer());
logger.debug(" Model: {}", gateway.getModel());
logger.debug(" Serial No: {}", gateway.getSerialNo());
logger.debug(" SIM IMSI: {}", gateway.getImsi());
logger.debug(" Signal Level: {}", gateway.getSignalLevel() + " dBm");
logger.debug(" Battery Level: {}", gateway.getBatteryLevel() + "%");
} catch (Exception e){
logger.error(e.getMessage(), e);
}
}
}
synchronized public void stop() {
try {
ready = false;
service.stopService();
} catch (SMSLibException|IOException|InterruptedException e) {
logger.error(e.getMessage(), e);
}
}
synchronized public boolean send(OutboundMessage msg) {
msg.setEncoding(Message.MessageEncodings.ENCUCS2); // 中文
msg.setStatusReport(true); // 发送状态报告
try {
outboundMessages.addFirst(msg);
logger.info("Send message: {}, {}", msg.getRefNo(), msg.getText());
boolean result = service.sendMessage(msg);
logger.info("Send message done");
return result;
} catch (TimeoutException |GatewayException|IOException|InterruptedException e) {
logger.error(e.getMessage(), e);
return false;
}
}
synchronized public boolean queue(OutboundMessage msg) {
AbstractQueueManager queueManager = service.getQueueManager();
for (AGateway gateway : service.getGateways()) {
int queueSize = queueManager.pendingQueueSize(gateway.getGatewayId());
logger.info("Queue size: {} {}", gateway.getGatewayId(), queueSize);
}
msg.setEncoding(Message.MessageEncodings.ENCUCS2); // 中文
msg.setStatusReport(true); // 发送状态报告
return service.queueMessage(msg);
}
public int countInboundMessages() {
return inboundMessages.size();
}
public List<InboundMessage> listInboundMessages(int offset, int limit) {
if (offset > inboundMessages.size() - 1)
offset = inboundMessages.size() - 1;
if (offset < 0)
offset = 0;
int to = offset + limit;
if (to < offset)
to = offset;
if (to > inboundMessages.size())
to = inboundMessages.size();
return inboundMessages.subList(offset, to);
}
public boolean isReady() {
return ready;
}
public int countOutboundMessages() {
return outboundMessages.size();
}
public List<OutboundMessage> listOutboundMessages(int offset, int limit) {
if (offset > outboundMessages.size() - 1)
offset = outboundMessages.size() - 1;
if (offset < 0)
offset = 0;
int to = offset + limit;
if (to < offset)
to = offset;
if (to > outboundMessages.size())
to = outboundMessages.size();
return outboundMessages.subList(offset, to);
}
synchronized public void purge() {
while (inboundMessages.size() > HISTORY_SIZE) {
inboundMessages.removeLast();
}
while (outboundMessages.size() > HISTORY_SIZE) {
outboundMessages.removeLast();
}
}
public static void main(String[] args) {
ModemService modemService = new ModemService();
modemService.init();
/*OutboundMessage msg = new OutboundMessage("13800138000", "English 中文短信内容, 中文短信内容，中文短信内容");
boolean result = modemService.send(msg);
logger.info("result {}", result);*/
logger.info("Now Sleeping - Hit <enter> to terminate.");
try {
System.in.read();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
logger.info("Now stopping");
modemService.stop();
}
}
对于inbound消息的处理, 要区分普通短信和到达通知消息. 为防止写满SIM卡, 在接收到消息后都会立即删除
public class InboundNotification implements IInboundMessageNotification {
private static final Logger logger = LoggerFactory.getLogger(InboundNotification.class);
private final List<InboundMessage> inboundMessages;
public InboundNotification(List<InboundMessage> inboundMessages) {
this.inboundMessages = inboundMessages;
}
@Override
public void process(AGateway gateway, MessageTypes msgType, InboundMessage msg) {
logger.info("Inbound notification from: {}", gateway.getGatewayId());
logger.info("messageType: {}", msgType);
logger.info("message: {}", msg);
inboundMessages.add(0, msg);
if (msg.getMemLocation().equals("SR")) {
try {
if (gateway.deleteMessage(msg)) {
logger.info("Deleted status report: {}", msg.getId());
} else {
logger.error("Failed to delete status report: {}", msg.getId());
}
} catch (TimeoutException|GatewayException|IOException|InterruptedException e) {
logger.error(e.getMessage(), e);
}
} else {
try {
if (gateway.deleteMessage(msg)) {
logger.info("Deleted message: {}", msg.getId());
} else {
logger.error("Failed to delete message: {}", msg.getId());
}
} catch (TimeoutException|GatewayException|IOException|InterruptedException e) {
logger.error(e.getMessage(), e);
}
}
}
}
使用中的几点观察:
1. 代码中如果不调用 service.stopService() 方法, main进程将一直堵塞. 在Windows 10下, 若未调用stopService()即退出Java应用, 可能导致再次启动失败.
2. 如果SIM卡未注册到移动网络(接触不良, 信号不好, 或欠费), 都会导致service.startService() 方法执行失败
3. startService() 和 stopService() 都会需要几秒的执行时间.
4. sendMessage()是实时发送, 但是接收短信在服务启动后有数分钟的延迟, 数条短信会在延迟七八分钟后一起到达, 在服务运行一段时间后, 接收短信延迟会变小.
本文使用的硬件
从节电考虑, 未使用普通的x86服务器或主机, 而是使用了一块Amlogic S905L, 1G RAM的R3300-L, 在上面用8G TF card运行了64位的Armbian Linux, 经实际测试, 运行稳定.
进一步了解Smslib的内部机制
对于单张卡每小时200条每日1000条的限制, 但是物理发送的频率为每分钟12次. 实际使用中, 群发任务会集中在每日的特定时段, 目标是在任务启动后不超出运营商限制的前提下尽快完成, 因此可能会要求单张卡用满每小时200条的频率另外在发送满1000条后标记为不可用. 在这个前提下, 假定群发的短信为10K条, 那么不同数量SIM卡并发的情况下最快执行时间为:
1: 9天5小时 2: 4天5小时 4: 2天3小时 9: 1天0.5小时 10: 4小时20分钟 16: 3小时3分钟 20: 2小时10分钟 32: 1小时12分钟 40: 1小时5分钟 100: 10分钟 128: 7分钟
在SIM卡数量较多时, 如何提高发送效率? 使用同步锁是不行的, 需要改进. 研究smslib的代码得到的信息:
1. AGateway有多个实现, ModemGateway只是其中的一种, 还有HTTPGateway, SMPPGateway. 所以使用smslib实际上是可以将http通道一块合并进来作为发送出口的.
Smslib内部实现了eztexting.com, bulksms.vsms.net, Kannel, Skype, clickatell.com这几家的短信接口, 如果使用自己的HTTP通道, 可以扩展HTTPGateway实现, 需要实现的方法为 startGateway(), stopGateway(), sendMessage(OutboundMessage msg), queryMessage(String refNo), queryCoverage(OutboundMessage msg), queryBalance()
2. Smslib的Service实现了群发机制
通过一个Collection<Group> groups的成员变量, 管理可用的手机号组, 可以通过接口addToGroup(), removeFromGroup()管理各个组的手机号, 在发送时, 如果OutboundMessage的recipient命中group名称, 则会将这个OutboundMessage展开为一个列表, 再通过gateway进行发送.
这个群发机制是支持多层递归的, 即一个group为aaa, 可以将aaa加入到bbb, 那么发送到bbb的消息, 会进一步将aaa里的手机号也加入进来.
3. 使用ModemGateway发送短信时, 其sendMessage方法内部是加了同步锁的
加锁代码为 synchronized(this.getDriver().getSYNCCommander()) { ... } 内部再区分是Protocols.PDU 还是 Protocols.TEXT 进行发送.
进一步查看ModemGateway这个类的代码, 其中大量使用了上面提到的同步锁, 因此外部代码加锁是不必要的, 可以去掉.
4. Service只有在STOPPED的状态下, 才能addGateway()和removeGateway(), 因此动态管理gateway是不行的. 在运行中, gateway可以start(), stop(), 但是默认的 RoundRobinLoadBalancer 在取gateway时并不会判断gateway的状态, 所以如果要动态分配gateway, 需要自己控制.
5. 短信发送后, 其到达状态的关联依据为refNo, 这个值在发送成功后会写回OutboundMessage对象.
可改进的方向
假定运行设备为32口的moden pool, pool根据当前的时间和历史发送记录, 定时计算当前可发的短信数量
对于每一个提交的发送任务, 先分解为最终的短信列表,
判断是否小于pool的当前可发短信数量, 若超出则返回失败, 否则继续
维护一个任务队列, 一个结果队列
每一个gateway关联一个executor, 实时判断当前是否可发短信(超限或状态为stopped), 如果不可则睡眠一个随机时间后再次判断, 如果可以则从短信队列中取出最早一条进行发送
如果短信队列为空, 则睡眠随机时间后再次执行gateway步骤
gateway发送一条短信完成后, 根据结果将短信放入结果列, 同时再次执行gateway步骤
根据配置是否失败重试, 以及重试次数上限, 将结果队列中状态为失败且可重试的短信再放入任务队列, 直至结果队列中再无需要再重试的短信
根据Inbound SR, 更新结果队列中的短信记录
定时清理结果队列中时间较早的记录
对任务提供接口回调, 以及查询接口