Tenlion/cm-cloud
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增socket，调整oauth认证

Browse Source
This commit is contained in:
wenc000 2020-07-26 23:19:49 +08:00
parent 9acc8cf704
commit 42301ca37d
15 changed files with 582 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cloud-common-plugin-oauth/src/main/java/com/cm/common/plugin/oauth/token/ClientTokenManager.java
View File

@ -50,7 +50,7 @@ public class ClientTokenManager {
params.put("client_id", oauthClientProperties.getClientId());
params.put("client_secret", oauthClientProperties.getClientSecret());
params.put("scope", "all");
String result = restTemplateUtil.doPostFormNormal(String.format("%s/oauth/token", oauthProperties.getOauthServer()), params);
String result = restTemplateUtil.doPostFormNormal(String.format("%s/oauth_client/token", oauthProperties.getOauthServer()), params);
if (StringUtils.isBlank(result)) {
LOG.error("客户端获取token失效");
} else {

23
cloud-common-socket-client/pom.xml Normal file
View File

@ -0,0 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>cm-cloud</artifactId>
<groupId>com.cm</groupId>
<version>1.0.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>cloud-common-socket-client</artifactId>
<dependencies>
<dependency>
<groupId>com.cm</groupId>
<artifactId>cloud-common-socket</artifactId>
<version>1.0.1-SNAPSHOT</version>
</dependency>
</dependencies>
</project>

111
cloud-common-socket-client/src/main/java/com/cm/socket/client/MessageSocketClient.java Normal file
View File

@ -0,0 +1,111 @@
package com.cm.socket.client;
import com.cm.socket.client.config.MessageSocketClientConfig;
import com.cm.socket.client.handler.MessageClientChannelInitializer;
import com.cm.socket.client.service.AbstractMessageClientService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
/**
* When you feel like quitting. Think about why you started
* 当你想要放弃的时候，想想当初你为何开始
*
* @ClassName: MessageSocketClient
* @Description: 消息客户端
* @Author: WangGeng
* @Date: 2020/7/5 10:03 下午
* @Version: 1.0
**/
public class MessageSocketClient implements Runnable {
private static final MessageSocketClient messageSocketClient = MessageSocketClientBuilder.messageSocketClient;
private AbstractMessageClientService abstractMessageClientService;
private MessageSocketClientConfig messageSocketClientConfig;
private int currentReconnectCount = 1;
private int currentReconnectTime = 1;
EventLoopGroup eventLoopGroup;
Bootstrap bootstrap;
private MessageSocketClient() {
this.eventLoopGroup = new NioEventLoopGroup();
this.bootstrap = new Bootstrap();
this.bootstrap.group(eventLoopGroup);
}
public static MessageSocketClient getInstance() {
return messageSocketClient;
}
public void initClient(AbstractMessageClientService abstractMessageClientService, MessageSocketClientConfig messageSocketClientConfig) {
this.messageSocketClientConfig = messageSocketClientConfig;
this.abstractMessageClientService = abstractMessageClientService;
this.bootstrap.channel(NioSocketChannel.class);
this.bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
this.bootstrap.handler(new MessageClientChannelInitializer(this.abstractMessageClientService, this.messageSocketClientConfig));
}
@Override
public void run() {
ChannelFuture channelFuture = this.bootstrap.connect(messageSocketClientConfig.getHost(), messageSocketClientConfig.getPort());
channelFuture.addListener((ChannelFutureListener) future -> {
if (!future.isSuccess()) {
future.channel().pipeline().fireChannelInactive();
}
});
channelFuture.channel().closeFuture();
}
private static class MessageSocketClientBuilder {
public static final MessageSocketClient messageSocketClient = new MessageSocketClient();
}
public static void main(String[] args) {
MessageSocketClientConfig messageSocketClientConfig = new MessageSocketClientConfig();
messageSocketClientConfig.setHost("127.0.0.1");
messageSocketClientConfig.setPort(9999);
messageSocketClientConfig.setDelayPingSeconds(3);
messageSocketClientConfig.setMaxReconnectCount(10);
messageSocketClientConfig.setReconnectTimeStep(1);
AbstractMessageClientService abstractMessageClientService = new AbstractMessageClientService() {
@Override
public void readMessage(String message) {
System.out.println(message);
}
};
MessageSocketClient messageSocketClient = MessageSocketClient.getInstance();
messageSocketClient.initClient(abstractMessageClientService, messageSocketClientConfig);
messageSocketClient.run();
new Thread(() -> {
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
abstractMessageClientService.sendMessage("我是客户端");
}).run();
}
public int getCurrentReconnectCount() {
return currentReconnectCount <= 1 ? 1 : currentReconnectCount;
}
public void setCurrentReconnectCount(int currentReconnectCount) {
this.currentReconnectCount = currentReconnectCount;
}
public int getCurrentReconnectTime() {
return currentReconnectTime <= 1 ? 1 : currentReconnectTime;
}
public void setCurrentReconnectTime(int currentReconnectTime) {
this.currentReconnectTime = currentReconnectTime;
}
}

80
cloud-common-socket-client/src/main/java/com/cm/socket/client/config/MessageSocketClientConfig.java Normal file
View File

@ -0,0 +1,80 @@
package com.cm.socket.client.config;
import org.springframework.stereotype.Component;
/**
* When you feel like quitting. Think about why you started
* 当你想要放弃的时候，想想当初你为何开始
*
* @ClassName: MessageSocketClientConfig
* @Description: 消息客户端配置
* @Author: WangGeng
* @Date: 2020/7/7 11:04 下午
* @Version: 1.0
**/
@Component
public class MessageSocketClientConfig {
private String host;
private int port;
private int maxReconnectCount;
private int reconnectTimeStep;
private int delayPingSeconds;
public String getHost() {
return host == null ? "" : host.trim();
}
public void setHost(String host) {
this.host = host;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public int getMaxReconnectCount() {
return maxReconnectCount <= 0 ? 20 : maxReconnectCount;
}
public void setMaxReconnectCount(int maxReconnectCount) {
this.maxReconnectCount = maxReconnectCount;
}
public int getReconnectTimeStep() {
return reconnectTimeStep <= 0 ? 0 : reconnectTimeStep;
}
public void setReconnectTimeStep(int reconnectTimeStep) {
this.reconnectTimeStep = reconnectTimeStep;
}
public int getDelayPingSeconds() {
return delayPingSeconds <= 0 ? 3 : delayPingSeconds;
}
public void setDelayPingSeconds(int delayPingSeconds) {
this.delayPingSeconds = delayPingSeconds;
}
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("{");
sb.append("\"host\":")
.append("\"").append(host).append("\"");
sb.append(",\"port\":")
.append(port);
sb.append(",\"maxReconnectCount\":")
.append(maxReconnectCount);
sb.append(",\"reconnectTimeStep\":")
.append(reconnectTimeStep);
sb.append(",\"delayPingSeconds\":")
.append(delayPingSeconds);
sb.append('}');
return sb.toString();
}
}

39
cloud-common-socket-client/src/main/java/com/cm/socket/client/handler/MessageClientChannelInitializer.java Normal file
View File

@ -0,0 +1,39 @@
package com.cm.socket.client.handler;
import com.cm.socket.client.MessageSocketClient;
import com.cm.socket.client.config.MessageSocketClientConfig;
import com.cm.socket.client.service.AbstractMessageClientService;
import com.cm.socket.decoder.MessageDecoder;
import com.cm.socket.encoder.MessageEncoder;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
/**
* When you feel like quitting. Think about why you started
* 当你想要放弃的时候，想想当初你为何开始
*
* @ClassName: MessageClientChannelInitializer
* @Description: 消息客户端初始化
* @Author: WangGeng
* @Date: 2020/7/6 9:32 上午
* @Version: 1.0
**/
public class MessageClientChannelInitializer extends ChannelInitializer<SocketChannel> {
private AbstractMessageClientService abstractMessageClientService;
private MessageSocketClientConfig messageSocketClientConfig;
public MessageClientChannelInitializer(AbstractMessageClientService abstractMessageClientService, MessageSocketClientConfig messageSocketClientConfig) {
this.abstractMessageClientService = abstractMessageClientService;
this.messageSocketClientConfig = messageSocketClientConfig;
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new MessageEncoder());
ch.pipeline().addLast(new MessageDecoder());
ch.pipeline().addLast(new MessageClientPingHandler(MessageSocketClient.getInstance(), messageSocketClientConfig));
ch.pipeline().addLast(new MessageClientHandler(abstractMessageClientService));
}
}

37
cloud-common-socket-client/src/main/java/com/cm/socket/client/handler/MessageClientHandler.java Normal file
View File

@ -0,0 +1,37 @@
package com.cm.socket.client.handler;
import com.cm.socket.client.service.AbstractMessageClientService;
import com.cm.socket.pojo.Message;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
/**
* When you feel like quitting. Think about why you started
* 当你想要放弃的时候，想想当初你为何开始
*
* @ClassName: MessageClientHandler
* @Description: 消息客户端处理器
* @Author: WangGeng
* @Date: 2020/7/6 9:34 上午
* @Version: 1.0
**/
public class MessageClientHandler extends SimpleChannelInboundHandler<Message> {
private AbstractMessageClientService abstractMessageClientService;
public MessageClientHandler(AbstractMessageClientService abstractMessageClientService) {
this.abstractMessageClientService = abstractMessageClientService;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
super.channelActive(ctx);
abstractMessageClientService.setChannelHandlerContext(ctx);
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
abstractMessageClientService.readMessage(msg.getContent());
}
}

104
cloud-common-socket-client/src/main/java/com/cm/socket/client/handler/MessageClientPingHandler.java Normal file
View File

@ -0,0 +1,104 @@
package com.cm.socket.client.handler;
import com.cm.socket.client.MessageSocketClient;
import com.cm.socket.client.config.MessageSocketClientConfig;
import com.cm.socket.enums.SocketMessageEnum;
import com.cm.socket.pojo.Message;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
/**
* When you feel like quitting. Think about why you started
* 当你想要放弃的时候，想想当初你为何开始
*
* @ClassName: MessageClientPingHandler
* @Description: 客户端ping
* @Author: WangGeng
* @Date: 2020/7/6 9:43 上午
* @Version: 1.0
**/
public class MessageClientPingHandler extends SimpleChannelInboundHandler<Message> {
private MessageSocketClientConfig messageSocketClientConfig;
private MessageSocketClient messageSocketClient;
public MessageClientPingHandler(MessageSocketClient messageSocketClient, MessageSocketClientConfig messageSocketClientConfig) {
this.messageSocketClient = messageSocketClient;
this.messageSocketClientConfig = messageSocketClientConfig;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("服务器连接成功");
messageSocketClient.setCurrentReconnectCount(1);
messageSocketClient.setCurrentReconnectTime(1);
super.channelActive(ctx);
ping(ctx.channel());
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
if (msg.getStart() != SocketMessageEnum.MESSAGE_TYPE_START.getType()) {
return;
}
if (msg.getType() == SocketMessageEnum.MESSAGE_TYPE_PONG.getType()) {
System.out.println("server pong");
return;
}
ctx.fireChannelRead(msg);
}
/**
* ping
*
* @param channel
*/
private void ping(Channel channel) {
ScheduledFuture<?> scheduledFuture = channel.eventLoop().schedule(() -> {
if (channel.isActive()) {
Message ping = new Message();
ping.setStart(SocketMessageEnum.MESSAGE_TYPE_START.getType());
ping.setType(SocketMessageEnum.MESSAGE_TYPE_PING.getType());
ping.setContent(SocketMessageEnum.MESSAGE_PING.getMessage());
channel.writeAndFlush(ping);
} else {
channel.closeFuture();
throw new RuntimeException();
}
}, messageSocketClientConfig.getDelayPingSeconds(), TimeUnit.SECONDS);
scheduledFuture.addListener(future -> {
if (future.isSuccess()) {
ping(channel);
}
});
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.err.println("连接断开");
int currentReconnectCount = messageSocketClient.getCurrentReconnectCount();
int waitTime = messageSocketClient.getCurrentReconnectTime();
if (messageSocketClient.getCurrentReconnectCount() > this.messageSocketClientConfig.getMaxReconnectCount()) {
System.out.println("服务器重连失败");
throw new RuntimeException();
}
System.out.println(waitTime + "s后进行第" + currentReconnectCount + "次重连");
ctx.channel().eventLoop().schedule(() -> {
MessageSocketClient.getInstance().run();
}, waitTime, TimeUnit.SECONDS);
messageSocketClient.setCurrentReconnectCount(++currentReconnectCount);
if (messageSocketClientConfig.getReconnectTimeStep() > 0) {
messageSocketClient.setCurrentReconnectTime(waitTime + messageSocketClientConfig.getReconnectTimeStep());
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

49
cloud-common-socket-client/src/main/java/com/cm/socket/client/service/AbstractMessageClientService.java Normal file
View File

@ -0,0 +1,49 @@
package com.cm.socket.client.service;
import com.cm.socket.enums.SocketMessageEnum;
import com.cm.socket.pojo.Message;
import io.netty.channel.ChannelHandlerContext;
import java.nio.channels.Channel;
/**
* When you feel like quitting. Think about why you started
* 当你想要放弃的时候，想想当初你为何开始
*
* @ClassName: IMessageClientService
* @Description: 客户端消息业务接口
* @Author: WangGeng
* @Date: 2020/7/6 10:50 上午
* @Version: 1.0
**/
public abstract class AbstractMessageClientService {
protected ChannelHandlerContext channelHandlerContext;
/**
* 发送消息
*
* @param sendMessage
*/
public void sendMessage(String sendMessage) {
if (sendMessage == null || sendMessage.isEmpty()) {
return;
}
Message message = new Message();
message.setStart(SocketMessageEnum.MESSAGE_TYPE_START.getType());
message.setType(SocketMessageEnum.MESSAGE_TYPE_MESSAGE.getType());
message.setContent(sendMessage);
channelHandlerContext.writeAndFlush(message);
}
/**
* 读取消息
*
* @param message
*/
public abstract void readMessage(String message);
public void setChannelHandlerContext(ChannelHandlerContext channelHandlerContext) {
this.channelHandlerContext = channelHandlerContext;
}
}

5
cloud-common-socket/pom.xml
View File

@ -12,6 +12,11 @@
<artifactId>cloud-common-socket</artifactId>
<dependencies>
<dependency>
<groupId>com.cm</groupId>
<artifactId>cloud-common</artifactId>
<version>1.0.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>

80
cloud-common-socket/src/main/java/com/cm/socket/Client.java
View File

@ -8,6 +8,14 @@ import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
/**
* When you feel like quitting. Think about why you started
@ -32,21 +40,65 @@ public class Client {
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("服务器连接失败");
super.channelInactive(ctx);
}
});
ch.pipeline().addLast(new MessageEncoder());
ch.pipeline().addLast(new MessageDecoder());
ch.pipeline().addLast(new SimpleChannelInboundHandler<Message>() {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
super.channelActive(ctx);
ping(ctx.channel());
}
private void ping(Channel channel) {
ScheduledFuture<?> schedule = channel.eventLoop().schedule(new Runnable() {
@Override
public void run() {
if (channel.isActive()) {
System.out.println("向服务器发送ping");
Message ping = new Message();
ping.setStart((byte) 0x01);
ping.setType((byte) 0x01);
ping.setContent("ping");
channel.writeAndFlush(ping);
} else {
System.err.println("连接断开");
channel.closeFuture();
throw new RuntimeException();
}
}
}, 3, TimeUnit.SECONDS);
schedule.addListener(new GenericFutureListener() {
@Override
public void operationComplete(Future future) throws Exception {
if (future.isSuccess()) {
ping(channel);
}
}
});
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
System.out.println(msg.getContent());
if(sendCount[0] < 4) {
Thread.sleep(5000);
Message message = new Message();
message.setStart((byte) 0x01);
message.setType((byte) 0x02);
message.setContent("你好服务器"+ sendCount[0] +"!");
ctx.channel().writeAndFlush(message);
sendCount[0]++;
}
// if (sendCount[0] < 4) {
// Thread.sleep(5000);
// Message message = new Message();
// message.setStart((byte) 0x01);
// message.setType((byte) 0x02);
// message.setContent("你好服务器" + sendCount[0] + "!");
// ctx.channel().writeAndFlush(message);
// sendCount[0]++;
// } else {
// ctx.close();
// }
}
@Override
@ -55,9 +107,17 @@ public class Client {
ctx.close();
}
});
}
});
ChannelFuture channelFuture = bootstrap.connect("localhost", 9999).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
future.channel().pipeline().fireChannelInactive();
}
}
});
ChannelFuture channelFuture = bootstrap.connect("localhost", 9999).sync();
Message message = new Message();
message.setStart((byte) 0x01);
message.setType((byte) 0x02);

15
cloud-common-socket/src/main/java/com/cm/socket/consts/ISocketConst.java Normal file
View File

@ -0,0 +1,15 @@
package com.cm.socket.consts;
/**
* When you feel like quitting. Think about why you started
* 当你想要放弃的时候，想想当初你为何开始
*
* @ClassName: ISocketConsts
* @Description: socket常量
* @Author: WangGeng
* @Date: 2020/7/6 9:15 上午
* @Version: 1.0
**/
public interface ISocketConst {
}

6
cloud-common-socket/src/main/java/com/cm/socket/decoder/MessageDecoder.java
View File

@ -4,6 +4,7 @@ import com.cm.socket.pojo.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.CharsetUtil;
import java.nio.charset.Charset;
import java.util.List;
@ -26,6 +27,9 @@ public class MessageDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if(in == null) {
return;
}
if (in.readableBytes() < HEADER_LENGTH) {
return;
}
@ -46,7 +50,7 @@ public class MessageDecoder extends ByteToMessageDecoder {
Message message = new Message();
message.setStart(start);
message.setType(type);
message.setContent(new String(contentBytes, Charset.forName("UTF-8")));
message.setContent(new String(contentBytes, CharsetUtil.UTF_8));
out.add(message);
}
}

3
cloud-common-socket/src/main/java/com/cm/socket/encoder/MessageEncoder.java
View File

@ -4,6 +4,7 @@ import com.cm.socket.pojo.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.util.CharsetUtil;
import java.nio.charset.Charset;
@ -21,7 +22,7 @@ public class MessageEncoder extends MessageToByteEncoder<Message> {
@Override
protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
byte[] contentBytes = msg.getContent().getBytes(Charset.forName("UTF-8"));
byte[] contentBytes = msg.getContent().getBytes(CharsetUtil.UTF_8);
out.writeByte(msg.getStart());
out.writeByte(msg.getType());
out.writeInt(contentBytes.length);

40
cloud-common-socket/src/main/java/com/cm/socket/enums/SocketMessageEnum.java Normal file
View File

@ -0,0 +1,40 @@
package com.cm.socket.enums;
/**
* When you feel like quitting. Think about why you started
* 当你想要放弃的时候，想想当初你为何开始
*
* @ClassName: ISocketMessageEnum
* @Description:
* @Author: WangGeng
* @Date: 2020/7/6 9:26 上午
* @Version: 1.0
**/
public enum SocketMessageEnum {
MESSAGE_TYPE_START((byte) (0xAA << 4)),
MESSAGE_TYPE_PING((byte) 0x01),
MESSAGE_TYPE_PONG((byte) 0x02),
MESSAGE_TYPE_MESSAGE((byte) 0x03),
MESSAGE_PING("PING"),
MESSAGE_PONG("PONG");
private byte type;
private String message;
SocketMessageEnum(byte type) {
this.type = type;
}
SocketMessageEnum(String message) {
this.message = message;
}
public byte getType() {
return type;
}
public String getMessage() {
return message == null ? "" : message.trim();
}
}

1
pom.xml
View File

@ -22,6 +22,7 @@
<module>cloud-common-plugin-sensitive</module>
<module>cloud-common-freemarker</module>
<module>cloud-common-socket</module>
<module>cloud-common-socket-client</module>
</modules>
<packaging>pom</packaging>
<description>成迈云</description>