Disruptor内存消息队列简单使用

墨蓝 2022-11-15 11:14 256阅读 0赞

## Disruptor内存消息队列 ##

最近在做一个有关使用内存消息队列到功能，比如将日志信息或点击统计信息持久化等操作，开始想着用java到内存队列作为缓冲区，后来在网上搜到Disruptor这个东西，神乎其神到，就简单了解了一下，做了一个demo，感觉还不错，可以用用，有关概念可以自行搜索，下面就简单介绍一下开发过程。

## 新建测试工程 ##

新建maven工程listen-disruptor

工程目录结构如下

![154858\_Sqx7\_39617.png][154858_Sqx7_39617.png]

pom.xml文件内容

1.  `<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"`
2.  `xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">`
3.  `<modelVersion>4.0.0</modelVersion>`
4.  `<groupId>listen</groupId>`
5.  `<artifactId>listen-disruptor</artifactId>`
6.  `<packaging>war</packaging>`
7.  `<version>0.0.1-SNAPSHOT</version>`
8.  `<name>listen-disruptor Maven Webapp</name>`
9.  `<url>http://maven.apache.org</url>`
10. `<dependencies>`
11. `<dependency>`
12. `<groupId>junit</groupId>`
13. `<artifactId>junit</artifactId>`
14. `<version>3.8.1</version>`
15. `<scope>test</scope>`
16. `</dependency>`
17.  
18. ``
19. `<dependency>`
20. `<groupId>com.lmax</groupId>`
21. `<artifactId>disruptor</artifactId>`
22. `<version>3.3.6</version>`
23. `</dependency>`
24. ``
25. `<dependency>`
26. `<groupId>com.alibaba</groupId>`
27. `<artifactId>fastjson</artifactId>`
28. `<version>1.2.20</version>`
29. `</dependency>`
30. `</dependencies>`
31. `<build>`
32. `<finalName>listen-disruptor</finalName>`
33. `</build>`
34. `</project>`

在Disruptor中，所有到生产和消费是基于事件驱动到，即事件对象。

## 编码 ##

新建事件对象类LogEvent

1.  `package com.listen.disruptor;`
2.   
3.  `import java.io.Serializable;`
4.  `import java.util.Date;`
5.   
6.  `import com.alibaba.fastjson.JSONObject;`
7.   
8.  `/**`
9.  `* 事件对象（日志事件）`
10. `* @author Administrator`
11. `*`
12. `*/`
13. `public class LogEvent implements Serializable {`
14.  
15. `private long logId;`
16. `private String content;`
17. `private Date date;`
18.  
19. `public LogEvent(){`
20.  
21. `}`
22.  
23. `public LogEvent(long logId, String content, Date date){`
24. `this.logId = logId;`
25. `this.content = content;`
26. `this.date = date;`
27. `}`
28.  
29. `public long getLogId() {`
30. `return logId;`
31. `}`
32.  
33. `public void setLogId(long logId) {`
34. `this.logId = logId;`
35. `}`
36.  
37. `public String getContent() {`
38. `return content;`
39. `}`
40.  
41. `public void setContent(String content) {`
42. `this.content = content;`
43. `}`
44.  
45. `public Date getDate() {`
46. `return date;`
47. `}`
48.  
49. `public void setDate(Date date) {`
50. `this.date = date;`
51. `}`
52.  
53. `public String toString(){`
54. `return JSONObject.toJSONString(this);`
55. `}`
56.  
57. `}`

事件工厂类，用来初始化预分配空到事件对象

1.  `package com.listen.disruptor;`
2.   
3.  `import com.lmax.disruptor.EventFactory;`
4.   
5.  `/**`
6.  `* 事件生成工厂（用来初始化预分配事件对象）`
7.  `* @author Administrator`
8.  `*`
9.  `*/`
10. `public class LogEventFactory implements EventFactory<LogEvent> {`
11.  
12. `public LogEvent newInstance() {`
13. `return new LogEvent();`
14. `}`
15.  
16. `}`

新建事件到消费者LogEventConsumer，这里演示只是做简单到打印显示

1.  `package com.listen.disruptor;`
2.   
3.  `import com.lmax.disruptor.EventHandler;`
4.   
5.  `/**`
6.  `* 事件到消费者`
7.  `* @author Administrator`
8.  `*`
9.  `*/`
10. `public class LogEventConsumer implements EventHandler<LogEvent> {`
11.  
12. `public void onEvent(LogEvent logEvent, long seq, boolean bool) throws Exception {`
13. `System.out.println("seq:" + seq + ",bool:" + bool + ",logEvent:" + logEvent.toString());`
14. `}`
15.  
16. `}`

新建事件的生产者LogEventProducer

1.  `package com.listen.disruptor;`
2.   
3.  `import java.util.Date;`
4.   
5.  `import com.lmax.disruptor.RingBuffer;`
6.   
7.  `public class LogEventProducer {`
8.   
9.  `private final RingBuffer<LogEvent> ringBuffer;`
10.  
11. `public LogEventProducer(RingBuffer<LogEvent> ringBuffer){`
12. `this.ringBuffer = ringBuffer;`
13. `}`
14.  
15. `public void onData(long logId, String content, Date date){`
16. `//ringBuffer类似一个队列，next就是下一个槽`
17. `long seq = ringBuffer.next();`
18. `try{`
19. `//用seq索引取出一个空到事件用于填充`
20. `LogEvent logEvent = ringBuffer.get(seq);`
21. `logEvent.setLogId(logId);`
22. `logEvent.setContent(content);`
23. `logEvent.setDate(date);`
24.  
25. `}`
26. `finally{`
27. `//最终发布事件，很重要`
28. `ringBuffer.publish(seq);`
29. `}`
30. `}`
31.  
32. `}`

至此有关消费和生产到代码已经码完，但是目前推荐另一种生产者到写法，两种都可以使用

新建LogEventProducerWithTranslator生产者

1.  `package com.listen.disruptor;`
2.   
3.  `import java.util.Date;`
4.   
5.  `import com.lmax.disruptor.EventTranslatorVararg;`
6.  `import com.lmax.disruptor.RingBuffer;`
7.   
8.  `/**`
9.  `* 使用translator方式到事件生产者发布事件`
10. `* @author Administrator`
11. `*`
12. `*/`
13. `public class LogEventProducerWithTranslator {`
14.  
15. `private final static EventTranslatorVararg<LogEvent> translator = new EventTranslatorVararg<LogEvent>() {`
16.  
17. `public void translateTo(LogEvent logEvent, long seq, Object... objs) {`
18. `logEvent.setLogId((Long) objs[0]);`
19. `logEvent.setContent((String) objs[1]);`
20. `logEvent.setDate((Date) objs[2]);`
21. `}`
22. `};`
23.  
24. `private final RingBuffer<LogEvent> ringBuffer;`
25. `public LogEventProducerWithTranslator(RingBuffer<LogEvent> ringBuffer){`
26. `this.ringBuffer = ringBuffer;`
27. `}`
28.  
29. `public void onData(long logId, String content, Date date){`
30. `this.ringBuffer.publishEvent(translator, logId, content, date);`
31. `}`
32. `}`

测试类LogEventMain

1.  `package com.listen.disruptor;`
2.   
3.  `import java.util.Date;`
4.  `import java.util.concurrent.Executor;`
5.  `import java.util.concurrent.Executors;`
6.  `import java.util.concurrent.ThreadFactory;`
7.   
8.  `import com.lmax.disruptor.RingBuffer;`
9.  `import com.lmax.disruptor.dsl.Disruptor;`
10.  
11. `public class LogEventMain {`
12.  
13. `public static void main(String[] args) {`
14. `// producer();`
15. `producerWithTranslator();`
16. `}`
17.  
18. `public static void producer(){`
19. `Executor executor = Executors.newCachedThreadPool();`
20. `LogEventFactory factory = new LogEventFactory();`
21. `int ringBufferSize = 1024;`
22. `Disruptor<LogEvent> disruptor = new Disruptor<LogEvent>(factory, ringBufferSize, executor);`
23. `disruptor.handleEventsWith(new LogEventConsumer());`
24. `disruptor.start();`
25. `RingBuffer<LogEvent> ringBuffer = disruptor.getRingBuffer();`
26.  
27. `LogEventProducer producer = new LogEventProducer(ringBuffer);`
28. `for(int i = 0; true; i++){`
29. `producer.onData(i, "c" + i, new Date());`
30. `}`
31. `}`
32.  
33. `public static void producerWithTranslator(){`
34. `LogEventFactory factory = new LogEventFactory();`
35. `int ringBufferSize = 1024;`
36. `ThreadFactory threadFactory = new ThreadFactory() {`
37.  
38. `public Thread newThread(Runnable r) {`
39. `return new Thread(r);`
40. `}`
41. `};`
42. `Disruptor<LogEvent> disruptor = new Disruptor<LogEvent>(factory, ringBufferSize, threadFactory);`
43. `disruptor.handleEventsWith(new LogEventConsumer());`
44. `disruptor.start();`
45. `RingBuffer<LogEvent> ringBuffer = disruptor.getRingBuffer();`
46. `LogEventProducerWithTranslator producer2 = new LogEventProducerWithTranslator(ringBuffer);`
47. `for(int i = 0; true; i++){`
48. `producer2.onData(i, "c" + i, new Date());`
49. `}`
50. `}`
51.  
52. `}`

测试类分别写类两种生产者到测试，其中LogEventProducerWithTranslator生产者中到translator变量还有其他几种写法，适用于其他到场景，用法都是类似到，可以自行研究一下

## 测试 ##

测试结果如下

1.  `seq:57804,bool:false,logEvent:{"content":"c57804","date":1477899755475,"logId":57804}`
2.  `seq:57805,bool:false,logEvent:{"content":"c57805","date":1477899755475,"logId":57805}`
3.  `seq:57806,bool:false,logEvent:{"content":"c57806","date":1477899755475,"logId":57806}`
4.  `seq:57807,bool:false,logEvent:{"content":"c57807","date":1477899755475,"logId":57807}`
5.  `seq:57808,bool:false,logEvent:{"content":"c57808","date":1477899755475,"logId":57808}`
6.  `seq:57809,bool:false,logEvent:{"content":"c57809","date":1477899755475,"logId":57809}`
7.  `seq:57810,bool:false,logEvent:{"content":"c57810","date":1477899755475,"logId":57810}`
8.  `seq:57811,bool:false,logEvent:{"content":"c57811","date":1477899755475,"logId":57811}`
9.  `seq:57812,bool:false,logEvent:{"content":"c57812","date":1477899755475,"logId":57812}`
10. `seq:57813,bool:false,logEvent:{"content":"c57813","date":1477899755475,"logId":57813}`
11. `seq:57814,bool:false,logEvent:{"content":"c57814","date":1477899755475,"logId":57814}`
12. `seq:57815,bool:false,logEvent:{"content":"c57815","date":1477899755475,"logId":57815}`
13. `seq:57816,bool:false,logEvent:{"content":"c57816","date":1477899755475,"logId":57816}`
14. `seq:57817,bool:false,logEvent:{"content":"c57817","date":1477899755475,"logId":57817}`
15. `seq:57818,bool:false,logEvent:{"content":"c57818","date":1477899755475,"logId":57818}`
16. `seq:57819,bool:false,logEvent:{"content":"c57819","date":1477899755475,"logId":57819}`
17. `seq:57820,bool:false,logEvent:{"content":"c57820","date":1477899755475,"logId":57820}`
18. `seq:57821,bool:false,logEvent:{"content":"c57821","date":1477899755475,"logId":57821}`
19. `seq:57822,bool:false,logEvent:{"content":"c57822","date":1477899755475,"logId":57822}`
20. `seq:57823,bool:false,logEvent:{"content":"c57823","date":1477899755475,"logId":57823}`
21. `seq:57824,bool:false,logEvent:{"content":"c57824","date":1477899755475,"logId":57824}`
22. `seq:57825,bool:false,logEvent:{"content":"c57825","date":1477899755475,"logId":57825}`

性能测试我也没有研究，毕竟也没有专业到测试机器，只是在自己到电脑上跑一下，完全满足业务需要。

码云地址：[https://git.oschina.net/git.listen/listen-disruptor][https_git.oschina.net_git.listen_listen-disruptor]

[154858_Sqx7_39617.png]: /images/20221022/63da3728917646e1995528005b739ce3.png
[https_git.oschina.net_git.listen_listen-disruptor]: https://git.oschina.net/git.listen/listen-disruptor