package com.ld.io.netty;
|
|
import com.ld.io.api.IoMsgConsumer;
|
import com.ld.io.api.IoSession;
|
|
import io.netty.util.concurrent.DefaultThreadFactory;
|
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
|
import java.util.concurrent.*;
|
|
/**
|
* 线程池 处理接收到的信息, 不占用netty的工作线程
|
*/
|
class ReceiveMessageThreadPool {
|
private static final String POOL_NAME = "netty-receive-message";
|
private final Logger logger = LoggerFactory.getLogger(getClass());
|
private static ExecutorService executorService;
|
private IoMsgConsumer messageConsumer;
|
|
public ReceiveMessageThreadPool(IoMsgConsumer messageConsumer) {
|
this.messageConsumer = messageConsumer;
|
}
|
|
static {
|
ThreadFactory threadFactory = new DefaultThreadFactory(POOL_NAME);
|
executorService = new ThreadPoolExecutor(100, 100, 0L,
|
TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(500),
|
threadFactory, new ThreadPoolExecutor.DiscardOldestPolicy());
|
}
|
|
void execute(IoSession session, byte[] bytes) {
|
executorService
|
.submit(() -> {
|
if (messageConsumer != null) {
|
String msg;
|
try {
|
//msg = new String(bytes, "UTF-8");
|
//logger.info("临时打印查看报文："+ msg);
|
messageConsumer.consume(session, bytes);
|
} catch (Exception e) {
|
try {
|
msg = new String(bytes, "UTF-8");
|
} catch (Exception e1) {
|
msg = "转换为字符串异常";
|
}
|
logger.error(
|
"Consume message happened error, business key="
|
+ session.getBusinessKey()
|
+ ", msg=" + msg, e);
|
}
|
} else {
|
logger.error("Message consumer not config! All request was ignored!");
|
}
|
});
|
}
|
}
|
