parent
333647d1cf
commit
389ff387ef
@ -0,0 +1,30 @@
|
||||
package com.baiye.client;
|
||||
|
||||
import com.baiye.disruptor.MessageConsumer;
|
||||
import com.baiye.dto.TranslatorData;
|
||||
import com.baiye.dto.TranslatorDataWapper;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.util.ReferenceCountUtil;
|
||||
|
||||
public class MessageConsumerImpl4Client extends MessageConsumer {
|
||||
|
||||
|
||||
public MessageConsumerImpl4Client(String consumerId) {
|
||||
super(consumerId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onEvent(TranslatorDataWapper wapper) throws Exception {
|
||||
TranslatorData response = wapper.getTranslatorData();
|
||||
ChannelHandlerContext ctx = wapper.getCtx();
|
||||
|
||||
try{
|
||||
// TODO: 2022/7/13 0013 这里替代为实际的业务处理
|
||||
System.out.println("Client端 :" + response);
|
||||
}finally {
|
||||
// 消息在buffer中,因此要进行释放 -> 方便进行GC
|
||||
// 最后一步如果有写就不需要考虑手动释放, 因为写的底层是用递归写完之后进行释放的
|
||||
ReferenceCountUtil.release(response);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,22 @@
|
||||
package com.baiye.disruptor;
|
||||
|
||||
import com.baiye.dto.TranslatorDataWapper;
|
||||
import com.lmax.disruptor.WorkHandler;
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* onEvent 方法不在这里进行实现 具体的消费逻辑给Netty去进行实现
|
||||
* @author q
|
||||
*/
|
||||
@Data
|
||||
public abstract class MessageConsumer implements WorkHandler<TranslatorDataWapper> {
|
||||
|
||||
/**
|
||||
* 消费者Id
|
||||
*/
|
||||
protected String consumerId;
|
||||
|
||||
public MessageConsumer(String consumerId) {
|
||||
this.consumerId = consumerId;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,32 @@
|
||||
package com.baiye.disruptor;
|
||||
|
||||
import com.baiye.dto.TranslatorData;
|
||||
import com.baiye.dto.TranslatorDataWapper;
|
||||
import com.lmax.disruptor.RingBuffer;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
|
||||
public class MessageProducer {
|
||||
|
||||
private String producerId;
|
||||
|
||||
private RingBuffer<TranslatorDataWapper> ringBuffer;
|
||||
|
||||
public MessageProducer(String producerId, RingBuffer<TranslatorDataWapper> ringBuffer) {
|
||||
this.producerId = producerId;
|
||||
this.ringBuffer = ringBuffer;
|
||||
}
|
||||
|
||||
public void onData(TranslatorData data, ChannelHandlerContext ctx) {
|
||||
long sequence = ringBuffer.next();
|
||||
|
||||
try{
|
||||
TranslatorDataWapper wapper = ringBuffer.get(sequence);
|
||||
wapper.setTranslatorData(data);
|
||||
wapper.setCtx(ctx);
|
||||
|
||||
}finally {
|
||||
ringBuffer.publish(sequence);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,144 @@
|
||||
package com.baiye.disruptor;
|
||||
|
||||
import com.baiye.dto.TranslatorDataWapper;
|
||||
import com.lmax.disruptor.*;
|
||||
import com.lmax.disruptor.dsl.ProducerType;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
/**
|
||||
* 环形缓冲池工厂类 - 单例模式
|
||||
*
|
||||
* @author q
|
||||
* @date 2022/07/13
|
||||
*/
|
||||
public class RingBufferWorkerPoolFactory {
|
||||
|
||||
|
||||
/**
|
||||
* 静态工厂
|
||||
*/
|
||||
public static class SingletonHolder{
|
||||
static final RingBufferWorkerPoolFactory instance = new RingBufferWorkerPoolFactory();
|
||||
}
|
||||
|
||||
private RingBufferWorkerPoolFactory(){
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* 对外调用
|
||||
* @return 返回工厂实例
|
||||
*/
|
||||
public static RingBufferWorkerPoolFactory getInstance() {
|
||||
return SingletonHolder.instance;
|
||||
}
|
||||
|
||||
/**
|
||||
* 生产者池 - 管理生产者
|
||||
*/
|
||||
private static Map<String, MessageProducer> producers = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* 消费者池 - 管理消费者
|
||||
*/
|
||||
private static Map<String, MessageConsumer> consumers = new ConcurrentHashMap<>();
|
||||
|
||||
/*
|
||||
定义 Disruptor 需要的元素
|
||||
*/
|
||||
private RingBuffer<TranslatorDataWapper> ringBuffer;
|
||||
|
||||
private SequenceBarrier sequenceBarrier;
|
||||
|
||||
private WorkerPool<TranslatorDataWapper> workerPool;
|
||||
|
||||
|
||||
/**
|
||||
* 初始化并启动
|
||||
*
|
||||
* @param producerType 生产者类型
|
||||
* @param bufferSize 缓冲区大小
|
||||
* @param waitStrategy 等待策略
|
||||
* @param messageConsumers 消息的消费者组
|
||||
*/
|
||||
public void initAndStart(ProducerType producerType, int bufferSize, WaitStrategy waitStrategy, MessageConsumer[] messageConsumers) {
|
||||
|
||||
// 1. 构建RingBuffer对象
|
||||
this.ringBuffer = RingBuffer.create(
|
||||
producerType,
|
||||
TranslatorDataWapper::new,
|
||||
bufferSize,
|
||||
waitStrategy);
|
||||
|
||||
// 2. 设置序号栅栏
|
||||
this.sequenceBarrier = this.ringBuffer.newBarrier();
|
||||
|
||||
// 3. 设置工作池
|
||||
this.workerPool = new WorkerPool<>(
|
||||
this.ringBuffer,
|
||||
this.sequenceBarrier,
|
||||
new EventExceptionHandler(),
|
||||
messageConsumers
|
||||
);
|
||||
|
||||
// 4. 把所构建的消费者放入池中
|
||||
for (MessageConsumer messageConsumer : messageConsumers) {
|
||||
consumers.put(messageConsumer.getConsumerId(), messageConsumer);
|
||||
}
|
||||
|
||||
// 5. 添加我们的 sequence 集合
|
||||
this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());
|
||||
|
||||
// 6. 启动我们的工作池 -> 这边一定要用自定义线程池
|
||||
// FIXME: 2022/7/13 0013 进行自定义线程池
|
||||
this.workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() / 2));
|
||||
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 获取 MessageProducer
|
||||
*
|
||||
* @param producerId MessageProducerId
|
||||
* @return {@link MessageProducer}
|
||||
*/
|
||||
public MessageProducer getMessageProducer(String producerId) {
|
||||
MessageProducer messageProducer = producers.get(producerId);
|
||||
if (null == messageProducer){
|
||||
messageProducer = new MessageProducer(producerId, this.ringBuffer);
|
||||
producers.put(producerId, messageProducer);
|
||||
}
|
||||
return messageProducer;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 事件异常处理程序
|
||||
*
|
||||
*
|
||||
*
|
||||
* @author q
|
||||
* @date 2022/07/13
|
||||
*/
|
||||
static class EventExceptionHandler implements ExceptionHandler<TranslatorDataWapper>{
|
||||
|
||||
@Override
|
||||
public void handleEventException(Throwable throwable, long l, TranslatorDataWapper translatorDataWapper) {
|
||||
// TODO: 2022/7/13 0013 todo...
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handleOnStartException(Throwable throwable) {
|
||||
// TODO: 2022/7/13 0013 todo...
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handleOnShutdownException(Throwable throwable) {
|
||||
// TODO: 2022/7/13 0013 todo...
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,22 @@
|
||||
package com.baiye.dto;
|
||||
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* 包装 Netty 使用的数据类型给 Disruptor 进行使用
|
||||
*
|
||||
* @author q
|
||||
* @date 2022/07/13
|
||||
*/
|
||||
@Data
|
||||
public class TranslatorDataWapper {
|
||||
|
||||
private TranslatorData translatorData;
|
||||
|
||||
private ChannelHandlerContext ctx;
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue