xie.bx
3 years ago
24 changed files with 357 additions and 168 deletions
@ -1,8 +0,0 @@ |
|||
package com.bx.imserver.constant; |
|||
|
|||
public class Constant { |
|||
|
|||
// public static String LOCAL_SERVER_ID = UUID.randomUUID().toString();
|
|||
|
|||
|
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
package com.bx.imserver.netty; |
|||
|
|||
public interface IMServer { |
|||
|
|||
boolean isReady(); |
|||
|
|||
void start(); |
|||
|
|||
void stop(); |
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
package com.bx.imserver.netty; |
|||
|
|||
import com.bx.imcommon.contant.RedisKey; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.CommandLineRunner; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import javax.annotation.PreDestroy; |
|||
import java.util.List; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
public class IMServerMap implements CommandLineRunner { |
|||
|
|||
public static volatile long serverId = 0; |
|||
|
|||
@Autowired |
|||
RedisTemplate<String,Object> redisTemplate; |
|||
|
|||
@Autowired |
|||
private List<IMServer> imServers; |
|||
|
|||
@Override |
|||
public void run(String... args) throws Exception { |
|||
// 初始化SERVER_ID
|
|||
String key = RedisKey.IM_MAX_SERVER_ID; |
|||
serverId = redisTemplate.opsForValue().increment(key,1); |
|||
// 启动服务
|
|||
for(IMServer imServer:imServers){ |
|||
imServer.start(); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy(){ |
|||
// 停止服务
|
|||
for(IMServer imServer:imServers){ |
|||
imServer.stop(); |
|||
} |
|||
} |
|||
} |
|||
@ -1,4 +1,4 @@ |
|||
package com.bx.imserver.processor; |
|||
package com.bx.imserver.netty.processor; |
|||
|
|||
|
|||
import io.netty.channel.ChannelHandlerContext; |
|||
@ -1,4 +1,4 @@ |
|||
package com.bx.imserver.processor; |
|||
package com.bx.imserver.netty.processor; |
|||
|
|||
import com.bx.imcommon.enums.IMCmdType; |
|||
import com.bx.imserver.util.SpringContextHolder; |
|||
@ -0,0 +1,94 @@ |
|||
package com.bx.imserver.netty.tcp; |
|||
|
|||
import com.bx.imserver.netty.IMChannelHandler; |
|||
import com.bx.imserver.netty.IMServer; |
|||
import com.bx.imserver.netty.tcp.endecode.MessageProtocolDecoder; |
|||
import com.bx.imserver.netty.tcp.endecode.MessageProtocolEncoder; |
|||
import io.netty.bootstrap.ServerBootstrap; |
|||
import io.netty.channel.*; |
|||
import io.netty.channel.nio.NioEventLoopGroup; |
|||
import io.netty.channel.socket.nio.NioServerSocketChannel; |
|||
import io.netty.handler.timeout.IdleStateHandler; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
|
|||
/** |
|||
* TCP服务器,用于连接非网页的客户端,协议格式： 4字节内容的长度+IMSendInfo的JSON序列化 |
|||
* |
|||
* @author Blue |
|||
* @date 2022-11-20 |
|||
*/ |
|||
@Slf4j |
|||
@Component |
|||
@ConditionalOnProperty(prefix = "tcpsocket", value = "enable", havingValue = "true",matchIfMissing = true) |
|||
public class TcpSocketServer implements IMServer { |
|||
|
|||
|
|||
private volatile boolean ready = false; |
|||
|
|||
@Value("${tcpsocket.port}") |
|||
private int port; |
|||
|
|||
private ServerBootstrap bootstrap = new ServerBootstrap(); |
|||
private EventLoopGroup bossGroup = new NioEventLoopGroup(); |
|||
private EventLoopGroup workGroup = new NioEventLoopGroup(); |
|||
|
|||
@Override |
|||
public boolean isReady() { |
|||
return ready; |
|||
} |
|||
|
|||
@Override |
|||
public void start() { |
|||
// 设置为主从线程模型
|
|||
bootstrap.group(bossGroup, workGroup) |
|||
// 设置服务端NIO通信类型
|
|||
.channel(NioServerSocketChannel.class) |
|||
// 设置ChannelPipeline，也就是业务职责链，由处理的Handler串联而成，由从线程池处理
|
|||
.childHandler(new ChannelInitializer<Channel>() { |
|||
// 添加处理的Handler，通常包括消息编解码、业务处理，也可以是日志、权限、过滤等
|
|||
@Override |
|||
protected void initChannel(Channel ch) throws Exception { |
|||
// 获取职责链
|
|||
ChannelPipeline pipeline = ch.pipeline(); |
|||
pipeline.addLast(new IdleStateHandler(15, 0, 0, TimeUnit.SECONDS)); |
|||
pipeline.addLast("encode",new MessageProtocolEncoder()); |
|||
pipeline.addLast("decode",new MessageProtocolDecoder()); |
|||
pipeline.addLast("handler", new IMChannelHandler()); |
|||
} |
|||
}) |
|||
// bootstrap 还可以设置TCP参数，根据需要可以分别设置主线程池和从线程池参数，来优化服务端性能。
|
|||
// 其中主线程池使用option方法来设置，从线程池使用childOption方法设置。
|
|||
// backlog表示主线程池中在套接口排队的最大数量，队列由未连接队列（三次握手未完成的）和已连接队列
|
|||
.option(ChannelOption.SO_BACKLOG, 5) |
|||
// 表示连接保活，相当于心跳机制，默认为7200s
|
|||
.childOption(ChannelOption.SO_KEEPALIVE, true); |
|||
|
|||
try { |
|||
// 绑定端口，启动select线程，轮询监听channel事件，监听到事件之后就会交给从线程池处理
|
|||
Channel channel = bootstrap.bind(port).sync().channel(); |
|||
// 就绪标志
|
|||
this.ready = true; |
|||
log.info("tcp server 初始化完成,端口：{}",port); |
|||
// 等待服务端口关闭
|
|||
//channel.closeFuture().sync();
|
|||
} catch (InterruptedException e) { |
|||
log.info("tcp server 初始化异常",e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void stop(){ |
|||
log.info("tcp server 停止"); |
|||
bossGroup.shutdownGracefully(); |
|||
workGroup.shutdownGracefully(); |
|||
this.ready = false; |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -0,0 +1,29 @@ |
|||
package com.bx.imserver.netty.tcp.endecode; |
|||
|
|||
import com.bx.imcommon.model.IMSendInfo; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import io.netty.buffer.ByteBuf; |
|||
import io.netty.channel.ChannelHandlerContext; |
|||
import io.netty.handler.codec.ReplayingDecoder; |
|||
import io.netty.util.CharsetUtil; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Slf4j |
|||
public class MessageProtocolDecoder extends ReplayingDecoder { |
|||
|
|||
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception { |
|||
if(byteBuf.readableBytes()< 4){ |
|||
return; |
|||
} |
|||
// 获取到包的长度
|
|||
long length=byteBuf.readLong(); |
|||
// 转成IMSendInfo
|
|||
ByteBuf contentBuf = byteBuf.readBytes((int)length); |
|||
String content = contentBuf.toString(CharsetUtil.UTF_8); |
|||
ObjectMapper objectMapper = new ObjectMapper(); |
|||
IMSendInfo sendInfo = objectMapper.readValue(content, IMSendInfo.class); |
|||
list.add(sendInfo); |
|||
} |
|||
} |
|||
@ -0,0 +1,22 @@ |
|||
package com.bx.imserver.netty.tcp.endecode; |
|||
|
|||
import com.bx.imcommon.model.IMSendInfo; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import io.netty.buffer.ByteBuf; |
|||
import io.netty.channel.ChannelHandlerContext; |
|||
import io.netty.handler.codec.MessageToByteEncoder; |
|||
|
|||
public class MessageProtocolEncoder extends MessageToByteEncoder<IMSendInfo> { |
|||
|
|||
@Override |
|||
protected void encode(ChannelHandlerContext channelHandlerContext, IMSendInfo sendInfo, ByteBuf byteBuf) throws Exception { |
|||
ObjectMapper objectMapper = new ObjectMapper(); |
|||
String content = objectMapper.writeValueAsString(sendInfo); |
|||
byte[] bytes = content.getBytes("UTF-8"); |
|||
// 写入长度
|
|||
byteBuf.writeLong(bytes.length); |
|||
// 写入命令体
|
|||
byteBuf.writeBytes(bytes); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,102 @@ |
|||
package com.bx.imserver.netty.ws; |
|||
|
|||
import com.bx.imserver.netty.IMChannelHandler; |
|||
import com.bx.imserver.netty.IMServer; |
|||
import com.bx.imserver.netty.ws.endecode.MessageProtocolDecoder; |
|||
import com.bx.imserver.netty.ws.endecode.MessageProtocolEncoder; |
|||
import io.netty.bootstrap.ServerBootstrap; |
|||
import io.netty.channel.*; |
|||
import io.netty.channel.nio.NioEventLoopGroup; |
|||
import io.netty.channel.socket.nio.NioServerSocketChannel; |
|||
import io.netty.handler.codec.http.HttpObjectAggregator; |
|||
import io.netty.handler.codec.http.HttpServerCodec; |
|||
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; |
|||
import io.netty.handler.stream.ChunkedWriteHandler; |
|||
import io.netty.handler.timeout.IdleStateHandler; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
|
|||
/** |
|||
* WS服务器,用于连接网页的客户端,协议格式: 直接IMSendInfo的JSON序列化 |
|||
* |
|||
* @author Blue |
|||
* @date 2022-11-20 |
|||
*/ |
|||
@Slf4j |
|||
@Component |
|||
@ConditionalOnProperty(prefix = "websocket", value = "enable", havingValue = "true",matchIfMissing = true) |
|||
public class WebSocketServer implements IMServer { |
|||
|
|||
@Value("${websocket.port}") |
|||
private int port; |
|||
|
|||
private volatile boolean ready = false; |
|||
|
|||
private ServerBootstrap bootstrap = new ServerBootstrap(); |
|||
private EventLoopGroup bossGroup = new NioEventLoopGroup(); |
|||
private EventLoopGroup workGroup = new NioEventLoopGroup(); |
|||
|
|||
|
|||
@Override |
|||
public boolean isReady(){ |
|||
return ready; |
|||
} |
|||
|
|||
@Override |
|||
public void start() { |
|||
// 设置为主从线程模型
|
|||
bootstrap.group(bossGroup, workGroup) |
|||
// 设置服务端NIO通信类型
|
|||
.channel(NioServerSocketChannel.class) |
|||
// 设置ChannelPipeline，也就是业务职责链，由处理的Handler串联而成，由从线程池处理
|
|||
.childHandler(new ChannelInitializer<Channel>() { |
|||
// 添加处理的Handler，通常包括消息编解码、业务处理，也可以是日志、权限、过滤等
|
|||
@Override |
|||
protected void initChannel(Channel ch) throws Exception { |
|||
// 获取职责链
|
|||
ChannelPipeline pipeline = ch.pipeline(); |
|||
pipeline.addLast(new IdleStateHandler(15, 0, 0, TimeUnit.SECONDS)); |
|||
pipeline.addLast("http-codec", new HttpServerCodec()); |
|||
pipeline.addLast("aggregator", new HttpObjectAggregator(65535)); |
|||
pipeline.addLast("http-chunked", new ChunkedWriteHandler()); |
|||
pipeline.addLast(new WebSocketServerProtocolHandler("/im")); |
|||
pipeline.addLast("encode",new MessageProtocolEncoder()); |
|||
pipeline.addLast("decode",new MessageProtocolDecoder()); |
|||
pipeline.addLast("handler", new IMChannelHandler()); |
|||
} |
|||
}) |
|||
// bootstrap 还可以设置TCP参数，根据需要可以分别设置主线程池和从线程池参数，来优化服务端性能。
|
|||
// 其中主线程池使用option方法来设置，从线程池使用childOption方法设置。
|
|||
// backlog表示主线程池中在套接口排队的最大数量，队列由未连接队列（三次握手未完成的）和已连接队列
|
|||
.option(ChannelOption.SO_BACKLOG, 5) |
|||
// 表示连接保活，相当于心跳机制，默认为7200s
|
|||
.childOption(ChannelOption.SO_KEEPALIVE, true); |
|||
|
|||
try { |
|||
// 绑定端口，启动select线程，轮询监听channel事件，监听到事件之后就会交给从线程池处理
|
|||
Channel channel = bootstrap.bind(port).sync().channel(); |
|||
// 就绪标志
|
|||
this.ready = true; |
|||
log.info("websocket server 初始化完成,端口：{}",port); |
|||
// 等待服务端口关闭
|
|||
//channel.closeFuture().sync();
|
|||
} catch (InterruptedException e) { |
|||
log.info("websocket server 初始化异常",e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void stop() { |
|||
log.info("websocket server 停止"); |
|||
bossGroup.shutdownGracefully(); |
|||
workGroup.shutdownGracefully(); |
|||
this.ready = false; |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -1,4 +1,4 @@ |
|||
package com.bx.imserver.ws.endecode; |
|||
package com.bx.imserver.netty.ws.endecode; |
|||
|
|||
import com.bx.imcommon.model.IMSendInfo; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
@ -1,4 +1,4 @@ |
|||
package com.bx.imserver.ws.endecode; |
|||
package com.bx.imserver.netty.ws.endecode; |
|||
|
|||
import com.bx.imcommon.model.IMSendInfo; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
@ -1,101 +0,0 @@ |
|||
package com.bx.imserver.ws; |
|||
|
|||
import com.bx.imcommon.contant.RedisKey; |
|||
import com.bx.imserver.ws.endecode.MessageProtocolDecoder; |
|||
import com.bx.imserver.ws.endecode.MessageProtocolEncoder; |
|||
import io.netty.bootstrap.ServerBootstrap; |
|||
import io.netty.channel.*; |
|||
import io.netty.channel.nio.NioEventLoopGroup; |
|||
import io.netty.channel.socket.nio.NioServerSocketChannel; |
|||
import io.netty.handler.codec.http.HttpObjectAggregator; |
|||
import io.netty.handler.codec.http.HttpServerCodec; |
|||
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; |
|||
import io.netty.handler.stream.ChunkedWriteHandler; |
|||
import io.netty.handler.timeout.IdleStateHandler; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
public class WebsocketServer { |
|||
|
|||
public static long serverId = 0; |
|||
|
|||
@Autowired |
|||
RedisTemplate<String,Object> redisTemplate; |
|||
|
|||
private volatile boolean ready = false; |
|||
|
|||
|
|||
@PostConstruct |
|||
public void init(){ |
|||
// 初始化SERVER_ID
|
|||
String key = RedisKey.IM_MAX_SERVER_ID; |
|||
serverId = redisTemplate.opsForValue().increment(key,1); |
|||
} |
|||
|
|||
public boolean isReady(){ |
|||
return ready; |
|||
} |
|||
|
|||
public long getServerId(){ |
|||
return serverId; |
|||
} |
|||
|
|||
public void start(int port) { |
|||
// 服务端启动辅助类，用于设置TCP相关参数
|
|||
ServerBootstrap bootstrap = new ServerBootstrap(); |
|||
// 获取Reactor线程池
|
|||
EventLoopGroup bossGroup = new NioEventLoopGroup(); |
|||
EventLoopGroup workGroup = new NioEventLoopGroup(); |
|||
// 设置为主从线程模型
|
|||
bootstrap.group(bossGroup, workGroup) |
|||
// 设置服务端NIO通信类型
|
|||
.channel(NioServerSocketChannel.class) |
|||
// 设置ChannelPipeline，也就是业务职责链，由处理的Handler串联而成，由从线程池处理
|
|||
.childHandler(new ChannelInitializer<Channel>() { |
|||
// 添加处理的Handler，通常包括消息编解码、业务处理，也可以是日志、权限、过滤等
|
|||
@Override |
|||
protected void initChannel(Channel ch) throws Exception { |
|||
// 获取职责链
|
|||
ChannelPipeline pipeline = ch.pipeline(); |
|||
pipeline.addLast(new IdleStateHandler(15, 0, 0, TimeUnit.SECONDS)); |
|||
pipeline.addLast("http-codec", new HttpServerCodec()); |
|||
pipeline.addLast("aggregator", new HttpObjectAggregator(65535)); |
|||
pipeline.addLast("http-chunked", new ChunkedWriteHandler()); |
|||
pipeline.addLast(new WebSocketServerProtocolHandler("/im")); |
|||
pipeline.addLast("encode",new MessageProtocolEncoder()); |
|||
pipeline.addLast("decode",new MessageProtocolDecoder()); |
|||
pipeline.addLast("handler", new WebSocketHandler()); |
|||
} |
|||
}) |
|||
// bootstrap 还可以设置TCP参数，根据需要可以分别设置主线程池和从线程池参数，来优化服务端性能。
|
|||
// 其中主线程池使用option方法来设置，从线程池使用childOption方法设置。
|
|||
// backlog表示主线程池中在套接口排队的最大数量，队列由未连接队列（三次握手未完成的）和已连接队列
|
|||
.option(ChannelOption.SO_BACKLOG, 5) |
|||
// 表示连接保活，相当于心跳机制，默认为7200s
|
|||
.childOption(ChannelOption.SO_KEEPALIVE, true); |
|||
|
|||
try { |
|||
// 绑定端口，启动select线程，轮询监听channel事件，监听到事件之后就会交给从线程池处理
|
|||
Channel channel = bootstrap.bind(port).sync().channel(); |
|||
// 就绪标志
|
|||
this.ready = true; |
|||
log.info("websocket server 初始化完成...."); |
|||
// 等待服务端口关闭
|
|||
channel.closeFuture().sync(); |
|||
} catch (InterruptedException e) { |
|||
e.printStackTrace(); |
|||
} finally { |
|||
// 优雅退出，释放线程池资源
|
|||
bossGroup.shutdownGracefully(); |
|||
workGroup.shutdownGracefully(); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -1,11 +1,16 @@ |
|||
server: |
|||
port: 8877 |
|||
|
|||
websocket: |
|||
port: 8878 |
|||
|
|||
spring: |
|||
redis: |
|||
host: 127.0.0.1 |
|||
port: 6379 |
|||
database: 1 |
|||
|
|||
websocket: |
|||
enable: true |
|||
port: 8878 |
|||
|
|||
tcpsocket: |
|||
enable: false # 暂时不开启 |
|||
port: 8879 |
|||
Loading…
Reference in new issue