作者 crossoverJie

:sparkles: Introducing new features.新增接口回调

package com.crossoverjie.cim.client.config;
import com.crossoverjie.cim.client.handle.MsgHandleCaller;
import com.crossoverjie.cim.common.constant.Constants;
import com.crossoverjie.cim.common.protocol.CIMRequestProto;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import okhttp3.OkHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
/**
* Function:bean 配置
... ... @@ -19,9 +23,18 @@ import java.util.concurrent.TimeUnit;
@Configuration
public class BeanConfig {
private final static Logger LOGGER = LoggerFactory.getLogger(BeanConfig.class);
@Value("${cim.user.id}")
private long userId;
@Value("${cim.callback.thread.queue.size}")
private int queueSize;
@Value("${cim.callback.thread.pool.size}")
private int poolSize;
/**
* 创建心跳单例
... ... @@ -52,4 +65,33 @@ public class BeanConfig {
return builder.build();
}
/**
* 创建回调线程池
* @return
*/
@Bean("callBackThreadPool")
public ThreadPoolExecutor buildCallerThread(){
BlockingQueue<Runnable> queue = new LinkedBlockingQueue(queueSize);
ThreadFactory product = new ThreadFactoryBuilder()
.setNameFormat("product-%d")
.setDaemon(true)
.build();
ThreadPoolExecutor productExecutor = new ThreadPoolExecutor(poolSize, poolSize, 1, TimeUnit.MILLISECONDS, queue,product);
return productExecutor ;
}
/**
* 回调 bean
* @return
*/
@Bean
public MsgHandleCaller buildCaller(){
MsgHandleCaller caller = new MsgHandleCaller(msg -> {
LOGGER.warn("caller msg=[{}]" ,msg);
}) ;
return caller ;
}
}
... ...
... ... @@ -11,6 +11,8 @@ import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ThreadPoolExecutor;
/**
* Function:
*
... ... @@ -23,7 +25,9 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
private final static Logger LOGGER = LoggerFactory.getLogger(CIMClientHandle.class);
private MsgHandleCaller caller = SpringBeanFactory.getBean(MsgHandleCaller.class);
private ThreadPoolExecutor threadPoolExecutor = SpringBeanFactory.getBean("callBackThreadPool",ThreadPoolExecutor.class) ;;
@Override
... ... @@ -58,9 +62,24 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
//从服务端收到消息时被调用
//LOGGER.info("客户端收到消息={}",in.toString(CharsetUtil.UTF_8)) ;
//回调消息
callBackMsg(responseProtocol.getResMsg());
LOGGER.info(responseProtocol.getResMsg());
}
/**
* 回调消息
* @param msg
*/
private void callBackMsg(String msg) {
threadPoolExecutor.execute(() -> {
caller.getMsgHandleListener().handle(msg);
});
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
//异常时断开连接
... ...
package com.crossoverjie.cim.client.handle;
import com.crossoverjie.cim.client.service.CustomMsgHandleListener;
/**
* Function:消息回调 bean
*
* @author crossoverJie
* Date: 2018/12/26 17:37
* @since JDK 1.8
*/
public class MsgHandleCaller {
/**
* 回调接口
*/
private CustomMsgHandleListener msgHandleListener ;
public MsgHandleCaller(CustomMsgHandleListener msgHandleListener) {
this.msgHandleListener = msgHandleListener;
}
public CustomMsgHandleListener getMsgHandleListener() {
return msgHandleListener;
}
public void setMsgHandleListener(CustomMsgHandleListener msgHandleListener) {
this.msgHandleListener = msgHandleListener;
}
}
... ...
package com.crossoverjie.cim.client.service;
/**
* Function: 自定义消息回调
*
* @author crossoverJie
* Date: 2018/12/26 17:24
* @since JDK 1.8
*/
public interface CustomMsgHandleListener {
/**
* 消息回调
* @param msg
*/
void handle(String msg);
}
... ...
... ... @@ -28,6 +28,10 @@ cim.server.route.request.url=http://45.78.28.220:8083/login
cim.user.id=1545574841528
cim.user.userName=zhangsan
# 回调线程队列大小
cim.callback.thread.queue.size = 2
# 回调线程池大小
cim.callback.thread.pool.size = 2
# 关闭健康检查权限
management.security.enabled=false
... ...