heqijun
3 months ago
parent
48364eb76b
commit
711e2a9d8e
@ -1,4 +1,4 @@
|
||||
package com.mashibing.smmgateway;
|
||||
package com.mashibing.smsgateway;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
@ -0,0 +1,29 @@
|
||||
package com.mashibing.smsgateway.config;
|
||||
|
||||
import org.springframework.amqp.core.AcknowledgeMode;
|
||||
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
|
||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
|
||||
|
||||
/**
|
||||
* @author heqijun
|
||||
* @ClassName: RabbitMQConfig
|
||||
* @Description: 配置针对指定消费者的配置可以用配置类的方式
|
||||
* @date 2025/6/13 16:04
|
||||
*/
|
||||
|
||||
//@Configuration
|
||||
public class RabbitMQConfig {
|
||||
|
||||
// @Bean
|
||||
public SimpleRabbitListenerContainerFactory gatewayContainerFactory(ConnectionFactory connectionFactory,
|
||||
SimpleRabbitListenerContainerFactoryConfigurer configurer) {
|
||||
SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory = new SimpleRabbitListenerContainerFactory();
|
||||
simpleRabbitListenerContainerFactory.setConcurrentConsumers(5);
|
||||
simpleRabbitListenerContainerFactory.setPrefetchCount(10);
|
||||
simpleRabbitListenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
|
||||
configurer.configure(simpleRabbitListenerContainerFactory, connectionFactory);
|
||||
return simpleRabbitListenerContainerFactory;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,36 @@
|
||||
package com.mashibing.smsgateway.netty4;
|
||||
|
||||
import com.mashibing.smsgateway.netty4.entity.CmppMessageHeader;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.MessageToByteEncoder;
|
||||
|
||||
/**
|
||||
* 编码
|
||||
*/
|
||||
public class CMPPEncoder extends MessageToByteEncoder<Object> {
|
||||
|
||||
public CMPPEncoder(){
|
||||
super(false);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
|
||||
if (msg instanceof byte[]){
|
||||
out.writeBytes((byte[])msg);
|
||||
}else if(msg instanceof Integer){
|
||||
out.writeInt((Integer)msg);
|
||||
}else if(msg instanceof Byte){
|
||||
out.writeByte((Byte)msg);
|
||||
}else if(msg instanceof Long){
|
||||
out.writeLong((Long)msg);
|
||||
}else if(msg instanceof String){
|
||||
out.writeBytes(((String)msg).getBytes("UTF-16BE"));
|
||||
}else if (msg instanceof Character){
|
||||
out.writeChar((Character)msg);
|
||||
}else if (msg instanceof CmppMessageHeader){
|
||||
CmppMessageHeader c=(CmppMessageHeader)msg;
|
||||
out.writeBytes(c.toByteArray());
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,45 @@
|
||||
package com.mashibing.smsgateway.netty4;
|
||||
|
||||
|
||||
import com.mashibing.smsgateway.netty4.entity.CmppActiveTest;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.ChannelInboundHandlerAdapter;
|
||||
import io.netty.handler.timeout.IdleState;
|
||||
import io.netty.handler.timeout.IdleStateEvent;
|
||||
|
||||
|
||||
/**
|
||||
* 心跳Handler
|
||||
*/
|
||||
public class HeartHandler extends ChannelInboundHandlerAdapter {
|
||||
|
||||
private NettyClient client;
|
||||
public HeartHandler(NettyClient client){
|
||||
this.client=client;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
||||
System.out.println("初始化创建连接。。。");
|
||||
super.channelActive(ctx);
|
||||
}
|
||||
@Override
|
||||
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
|
||||
if (evt instanceof IdleStateEvent) {
|
||||
IdleStateEvent event = (IdleStateEvent) evt;
|
||||
IdleState state = event.state();
|
||||
if (state == IdleState.WRITER_IDLE || state == IdleState.ALL_IDLE) {
|
||||
client.submit(new CmppActiveTest());
|
||||
System.out.println("心跳启动!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
|
||||
}
|
||||
} else {
|
||||
super.userEventTriggered(ctx, evt);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
||||
super.channelInactive(ctx);
|
||||
client.reConnect(10);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,136 @@
|
||||
package com.mashibing.smsgateway.netty4;
|
||||
|
||||
import com.mashibing.smsgateway.netty4.entity.CmppConnect;
|
||||
import com.mashibing.smsgateway.netty4.entity.CmppMessageHeader;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.*;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
import io.netty.handler.logging.LoggingHandler;
|
||||
|
||||
|
||||
/**
|
||||
* 封装的netty客户端
|
||||
*/
|
||||
public class NettyClient {
|
||||
|
||||
// 通道
|
||||
private Channel channel;
|
||||
// IP
|
||||
private String host;
|
||||
// 端口
|
||||
private int port;
|
||||
// 用户名
|
||||
private String serviceId;
|
||||
// 密码
|
||||
private String pwd;
|
||||
|
||||
// 构建NettyClient
|
||||
public NettyClient(String host, int port, String serviceId, String pwd) {
|
||||
this.host = host;
|
||||
this.port = port;
|
||||
this.serviceId = serviceId;
|
||||
this.pwd = pwd;
|
||||
}
|
||||
|
||||
// 开启连接
|
||||
public void start() {
|
||||
try {
|
||||
doConnect(host, port);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
// 在保持连接的状态下提交信息
|
||||
public boolean submit(CmppMessageHeader submit) {
|
||||
if (isActive()) {
|
||||
channel.writeAndFlush(submit);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// 是否保持连接
|
||||
public boolean isActive() {
|
||||
if (channel == null) {
|
||||
return false;
|
||||
}
|
||||
if (!channel.isOpen() || !channel.isActive() || !channel.isWritable()) {
|
||||
//channel没开 或 没激活
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 重连机制 在 channelInactive 触发时调用
|
||||
*
|
||||
* @param reConnect
|
||||
*/
|
||||
public void reConnect(int reConnect) {
|
||||
int times = 0;
|
||||
while (true && times < reConnect) {
|
||||
try {
|
||||
if (!isActive()) {
|
||||
start();
|
||||
} else {
|
||||
try {
|
||||
Thread.sleep(10 * 1000);
|
||||
times++;
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
System.out.println("尝试重连...:" + host + ":" + port + " / " + serviceId);
|
||||
try {
|
||||
Thread.sleep(10 * 1000);
|
||||
times++;
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 建立连接
|
||||
* @param host
|
||||
* @param port
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public void doConnect(final String host, final int port) throws InterruptedException {
|
||||
|
||||
//创建线程组 - 手动设置线程数,默认为cpu核心数2倍
|
||||
EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);
|
||||
//创建引导程序
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
//保持长连接
|
||||
bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
|
||||
//将线程加入bootstrap
|
||||
bootstrap.group(eventLoopGroup)
|
||||
.remoteAddress(host, port)
|
||||
//使用指定通道类
|
||||
.channel(NioSocketChannel.class)
|
||||
//设置日志
|
||||
.handler(new LoggingHandler(LogLevel.INFO))
|
||||
//重写通道初始化方法
|
||||
.handler(new NettyClientInitializer(this));
|
||||
|
||||
ChannelFuture channelFuture = bootstrap.connect().sync();
|
||||
|
||||
channel = channelFuture.channel();
|
||||
|
||||
//账号登陆
|
||||
CmppMessageHeader cmppConnect = new CmppConnect(serviceId, pwd);
|
||||
channel.writeAndFlush(cmppConnect);
|
||||
|
||||
channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
|
||||
|
||||
//关闭前阻塞
|
||||
// channelFuture.channel().closeFuture().sync();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
package com.mashibing.smsgateway.netty4;
|
||||
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
|
||||
@Configuration
|
||||
public class NettyStartCMPP {
|
||||
|
||||
//ip
|
||||
public static String host = "127.0.0.1";
|
||||
//端口
|
||||
public static int port = 7890;
|
||||
//账号
|
||||
public static String serviceId = "laozheng";
|
||||
//密码
|
||||
public static String pwd = "JavaLaoZheng123!";
|
||||
|
||||
|
||||
|
||||
@Bean(initMethod = "start")
|
||||
public NettyClient nettyClient() {
|
||||
NettyClient cmppClient = new NettyClient(host, port, serviceId, pwd);
|
||||
return cmppClient;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
package com.mashibing.smsgateway.netty4.entity;
|
||||
|
||||
import com.mashibing.smsgateway.netty4.utils.Command;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.Unpooled;
|
||||
|
||||
public class CmppActiveTest extends CmppMessageHeader {
|
||||
|
||||
public CmppActiveTest() {
|
||||
super(Command.CMPP_ACTIVE_TEST, Command.CMPP2_VERSION);
|
||||
}
|
||||
|
||||
/**
|
||||
* 实现类必须自定义对象序列化
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
@Override
|
||||
public byte[] toByteArray() {
|
||||
ByteBuf buf = Unpooled.buffer(12);
|
||||
buf.writeInt(12);
|
||||
buf.writeInt(Command.CMPP_ACTIVE_TEST);
|
||||
buf.writeInt(0);
|
||||
return buf.array();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
package com.mashibing.smsgateway.netty4.entity;
|
||||
|
||||
import com.mashibing.smsgateway.netty4.utils.Command;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.Unpooled;
|
||||
|
||||
|
||||
public class CmppActiveTestResp extends CmppMessageHeader {
|
||||
public CmppActiveTestResp() {
|
||||
super(Command.CMPP_ACTIVE_TEST_RESP, Command.CMPP2_VERSION);
|
||||
}
|
||||
|
||||
/**
|
||||
* 实现类必须自定义对象序列化
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
@Override
|
||||
public byte[] toByteArray() {
|
||||
ByteBuf buf = Unpooled.buffer(4 + 4 + 4 + 1);
|
||||
buf.writeInt(4 + 4 + 4 + 1);
|
||||
buf.writeInt(Command.CMPP_ACTIVE_TEST_RESP);
|
||||
buf.writeInt(0);
|
||||
buf.writeByte(0);
|
||||
return buf.array();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,44 @@
|
||||
package com.mashibing.smsgateway.netty4.entity;
|
||||
|
||||
import com.mashibing.smsgateway.netty4.utils.Command;
|
||||
import com.mashibing.smsgateway.netty4.utils.MsgUtils;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.Unpooled;
|
||||
|
||||
/**
|
||||
* 与CMPP建立连接需要的信息
|
||||
*/
|
||||
public class CmppConnect extends CmppMessageHeader {
|
||||
|
||||
private String serviceId;
|
||||
|
||||
private String pwd;
|
||||
|
||||
public CmppConnect(String serviceId, String pwd) {
|
||||
super(Command.CMPP_CONNECT, Command.CMPP2_VERSION);
|
||||
this.serviceId = serviceId;
|
||||
this.pwd = pwd;
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] toByteArray() {
|
||||
ByteBuf buf = Unpooled.buffer(4 + 4 + 4 + 6 + 16 + 1 + 4);
|
||||
|
||||
//Total_Length
|
||||
buf.writeInt(4 + 4 + 4 + 6 + 16 + 1 + 4);
|
||||
//Command_Id
|
||||
buf.writeInt(Command.CMPP_CONNECT);
|
||||
//Sequence_Id
|
||||
buf.writeInt(MsgUtils.getSequence());
|
||||
//sourceAddr
|
||||
buf.writeBytes(MsgUtils.getLenBytes(serviceId, 6));
|
||||
//authenticatorSource
|
||||
buf.writeBytes(MsgUtils.getAuthenticatorSource(serviceId, pwd));
|
||||
//version
|
||||
buf.writeByte(1);
|
||||
//timestamp
|
||||
buf.writeInt(Integer.parseInt(MsgUtils.getTimestamp()));
|
||||
|
||||
return buf.array();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,38 @@
|
||||
package com.mashibing.smsgateway.netty4.entity;
|
||||
|
||||
|
||||
import com.mashibing.smsgateway.netty4.utils.MsgUtils;
|
||||
import org.apache.commons.lang3.ArrayUtils;
|
||||
|
||||
|
||||
/**
|
||||
* CMPP响应信息
|
||||
*/
|
||||
public class CmppSubmitResp {
|
||||
|
||||
private int sequenceId;
|
||||
|
||||
private int result;
|
||||
|
||||
private long msgId;
|
||||
|
||||
public CmppSubmitResp(byte[] bytes) {
|
||||
this.sequenceId = MsgUtils.bytesToInt(ArrayUtils.subarray(bytes, 8, 12));
|
||||
this.msgId = MsgUtils.bytesToLong(ArrayUtils.subarray(bytes, 12, 20));
|
||||
this.msgId = Math.abs(this.msgId);
|
||||
this.result = bytes[20];
|
||||
}
|
||||
|
||||
public int getResult() {
|
||||
return result;
|
||||
}
|
||||
|
||||
public long getMsgId() {
|
||||
return msgId;
|
||||
}
|
||||
|
||||
public int getSequenceId() {
|
||||
return sequenceId;
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in new issue