歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

內容：所有原創文章分類彙總及配套原始碼，涉及Java、Docker、Kubernetes、DevOPS等；

關於disruptor

disruptor是LMAX公司開發的一個高效能佇列，其作用和阻塞佇列(BlockingQueue)類似，都是在相同程序內、不同執行緒間傳遞資料(例如訊息、事件)，另外disruptor也有自己的一些特色：

  1. 以廣播的形式釋出事件，並且消費者之間存在依賴關係；
  2. 為事件提前分配記憶體；
  3. 無鎖演算法；

關於Ring Buffer(環形佇列)

  • 提到disruptor一般都會提到Ring Buffer(環形佇列)是它的特點，實際上從3.0版本之後，環形佇列只是用來儲存和更新事件資料，在其他更復雜的場景下，使用者可以通過自定義操作將其替換掉；

  • 簡單的說，disruptor官方認為Ring Buffe是核心概念（Core Concepts），但不是特色( key features)

本篇概覽

作為《disruptor筆記》系列的開篇，本篇有兩個任務：

  • 建立名為disruptor-tutorials的gradle工程，作為整個系列的父工程，該系列所有程式碼都是這個父工程下的module；

  • 在disruptor-tutorials下面新建名為basic-event的module，這是個springboot應用，作用是使用disruptor的基本功能：一個執行緒釋出事件，另一個執行緒消費事件，也就是對環形佇列最基本的操作，如下圖：

用disruptor實現訊息的釋出和消費的套路

  • 咱們提前小結用disruptor實現訊息的釋出和消費的套路，後面的開發按部就班即可，括號中是本篇對應的java類：
  1. 事件的定義：一個普通的bean（StringEvent.java）
  2. 事件工廠：定義如何生產事件的記憶體例項，這個例項剛從記憶體中建立，還沒有任何業務資料（StringEventFactory.java）
  3. 事件處理：封裝了消費單個事件的具體邏輯（StringEventHandler.java）
  4. 事件生產者：定義瞭如何將業務資料設定到還沒有業務資料的事件中，就是工廠創建出來的那種（StringEventProducer.java）
  5. 初始化邏輯：建立和啟動disruptor物件，將事件工廠傳給disruptor，建立事件生產者和事件處理物件，並分別與disruptor物件關聯；
  6. 業務邏輯：也就是呼叫事件生產者的onData方法釋出事件，本文的做法是在單元測試類中釋出事件，然後檢查消費的事件數和生產的事件數是否一致；

    7

環境資訊

《Disruptor筆記》系列涉及的環境資訊如下：

  1. 作業系統：64位win10
  2. JDK：1.8.0_281
  3. IDE：IntelliJ IDEA 2021.1.1 (Ultimate Edition)
  4. gradle：6.7.1
  5. springboot：2.3.8.RELEASE
  6. disruptor：3.4.4

原始碼下載

名稱 連結 備註
專案主頁 https://github.com/zq2599/blog_demos 該專案在GitHub上的主頁
git倉庫地址(https) https://github.com/zq2599/blog_demos.git 該專案原始碼的倉庫地址，https協議
git倉庫地址(ssh) git@github.com:zq2599/blog_demos.git 該專案原始碼的倉庫地址，ssh協議
  • 這個git專案中有多個資料夾，本次實戰的原始碼在disruptor-tutorials資料夾下，如下圖紅框所示：

建立父工程

  • 因為是系列文章，所以這裡做個父工程來管理所有依賴庫和外掛，新建名為disruptor-tutorials的gradle工程，build.gradle如下：
  1. import java.time.OffsetDateTime
  2. import java.time.format.DateTimeFormatter
  4. buildscript {
  5.     repositories {
  6.         maven {
  7.             url 'https://plugins.gradle.org/m2/'
  8.         }
  9.         // 如果有私服就在此配置，如果沒有請註釋掉
  10.         maven {
  11.             url 'http://192.168.50.43:8081/repository/aliyun-proxy/'
  12.         }
  13.         // 阿里雲
  14.         maven {
  15.             url 'http://maven.aliyun.com/nexus/content/groups/public/'
  16.         }
  18.         mavenCentral()
  19.     }
  20.     ext {
  21.         // 專案版本
  22.         projectVersion = '1.0-SNAPSHOT'
  24.         // sprignboot版本 https://github.com/spring-projects/spring-boot/releases
  25.         springBootVersion = '2.3.8.RELEASE'
  26.     }
  27. }
  29. plugins {
  30.     id 'java'
  31.     id 'java-library'
  32.     id 'org.springframework.boot' version "${springBootVersion}" apply false
  33.     id 'io.spring.dependency-management' version '1.0.11.RELEASE'
  34.     id 'net.nemerosa.versioning' version '2.14.0'
  35.     id 'io.franzbecker.gradle-lombok' version '4.0.0' apply false
  36.     id 'com.github.ben-manes.versions' version '0.36.0' // gradle dependencyUpdates
  37. }
  39. // If you attempt to build without the `--scan` parameter in `gradle 6.0+` it will cause a build error that it can't find
  40. // a buildScan property to change. This avoids that problem.
  41. if (hasProperty('buildScan')) {
  42.     buildScan {
  43.         termsOfServiceUrl = 'https://gradle.com/terms-of-service'
  44.         termsOfServiceAgree = 'yes'
  45.     }
  46. }
  48. wrapper {
  49.     gradleVersion = '6.7.1'
  50. }
  52. def buildTimeAndDate = OffsetDateTime.now()
  54. ext {
  55.     // 構建時取得當前日期和時間
  56.     buildDate = DateTimeFormatter.ISO_LOCAL_DATE.format(buildTimeAndDate)
  57.     buildTime = DateTimeFormatter.ofPattern('HH:mm:ss.SSSZ').format(buildTimeAndDate)
  58.     buildRevision = versioning.info.commit
  59. }
  61. allprojects {
  62.     apply plugin: 'java'
  63.     apply plugin: 'idea'
  64.     apply plugin: 'eclipse'
  65.     apply plugin: 'io.spring.dependency-management'
  66.     apply plugin: 'io.franzbecker.gradle-lombok'
  68.     compileJava {
  69.         sourceCompatibility = JavaVersion.VERSION_1_8
  70.         targetCompatibility = JavaVersion.VERSION_1_8
  71.         options.encoding = 'UTF-8'
  72.     }
  74.     compileJava.options*.compilerArgs = [
  75.             '-Xlint:all', '-Xlint:-processing'
  76.     ]
  78.     // Copy LICENSE
  79.     tasks.withType(Jar) {
  80.         from(project.rootDir) {
  81.             include 'LICENSE'
  82.             into 'META-INF'
  83.         }
  84.     }
  86.     // 寫入到MANIFEST.MF中的內容
  87.     jar {
  88.         manifest {
  89.             attributes(
  90.                     'Created-By': "${System.properties['java.version']} (${System.properties['java.vendor']} ${System.properties['java.vm.version']})".toString(),
  91.                     'Built-By': 'travis',
  92.                     'Build-Date': buildDate,
  93.                     'Build-Time': buildTime,
  94.                     'Built-OS': "${System.properties['os.name']}",
  95.                     'Build-Revision': buildRevision,
  96.                     'Specification-Title': project.name,
  97.                     'Specification-Version': projectVersion,
  98.                     'Specification-Vendor': 'Will Zhao',
  99.                     'Implementation-Title': project.name,
  100.                     'Implementation-Version': projectVersion,
  101.                     'Implementation-Vendor': 'Will Zhao'
  102.             )
  103.         }
  104.     }
  106.     repositories {
  107.         mavenCentral()
  109.         // 如果有私服就在此配置，如果沒有請註釋掉
  110.         maven {
  111.             url 'http://192.168.50.43:8081/repository/aliyun-proxy/'
  112.         }
  114.         // 阿里雲
  115.         maven {
  116.             url 'http://maven.aliyun.com/nexus/content/groups/public/'
  117.         }
  119.         jcenter()
  120.     }
  122.     buildscript {
  123.         repositories {
  124.             maven { url 'https://plugins.gradle.org/m2/' }
  125.         }
  126.     }
  127. }
  129. allprojects { project ->
  130.     buildscript {
  131.         dependencyManagement {
  132.             imports {
  133.                 mavenBom "org.springframework.boot:spring-boot-starter-parent:${springBootVersion}"
  134.                 mavenBom "org.junit:junit-bom:5.7.0"
  135.             }
  137.             dependencies {
  138.                 dependency 'org.projectlombok:lombok:1.16.16'
  139.                 dependency 'org.apache.commons:commons-lang3:3.11'
  140.                 dependency 'commons-collections:commons-collections:3.2.2'
  141.                 dependency 'com.lmax:disruptor:3.4.4'
  142.             }
  143.         }
  145.         ext {
  146.             springFrameworkVersion = dependencyManagement.importedProperties['spring-framework.version']
  147.         }
  148.     }
  149. }
  151. group = 'bolingcavalry'
  152. version = projectVersion
  • 接下來編寫訊息釋出和消費的程式碼；

新建module

  • 前面新建了整個《Disruptor筆記》系列的父工程，現在新建名為basic-event的module，其build.gradle內容如下：
  1. plugins {
  2.     id 'org.springframework.boot'
  3. }
  5. dependencies {
  6.     implementation 'org.projectlombok:lombok'
  7.     implementation 'org.springframework.boot:spring-boot-starter'
  8.     implementation 'org.springframework.boot:spring-boot-starter-web'
  9.     implementation 'com.lmax:disruptor'
  11.     testImplementation('org.springframework.boot:spring-boot-starter-test')
  12. }
  • 這個module是個springboot應用，啟動類如下：
  1. package com.bolingcavalry;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  6. @SpringBootApplication
  7. public class BasicEventApplication {
  8. 	public static void main(String[] args) {
  9. 		SpringApplication.run(BasicEventApplication.class, args);
  10. 	}
  11. }
  • 接下來按照前面總結的套路行事；

事件的定義

  • 事件定義類StringEvent.java，可見就是個普普通通的java bean：
  1. package com.bolingcavalry.service;
  3. import lombok.Data;
  4. import lombok.NoArgsConstructor;
  5. import lombok.ToString;
  7. @Data
  8. @ToString
  9. @NoArgsConstructor
  10. public class StringEvent {
  12.     private String value;
  13. }

事件工廠

  • 事件工廠的作用，是讓disruptor知道如何在記憶體中建立一個事件例項，不過，該例項和業務還沒有任何關係，本篇的事件工廠如下，可見就是建立StringEvent例項，並沒有特別的操作：
  1. package com.bolingcavalry.service;
  3. import com.lmax.disruptor.EventFactory;
  5. public class StringEventFactory implements EventFactory<StringEvent> {
  7.     @Override
  8.     public StringEvent newInstance() {
  9.         return new StringEvent();
  10.     }
  11. }

事件處理

  • 時間處理類的作用是定義一個事件如何被消費，裡面是具體的業務程式碼，每個事件都會執行此類的onEvent方法；
  • 本篇的事件處理類做的事情是列印事件內容，再用sleep消耗100毫秒，然後再呼叫外部傳入的Consumer實現類的accept方法：
  1. package com.bolingcavalry.service;
  3. import com.lmax.disruptor.EventHandler;
  4. import lombok.Setter;
  5. import lombok.extern.slf4j.Slf4j;
  6. import java.util.function.Consumer;
  8. @Slf4j
  9. public class StringEventHandler implements EventHandler<StringEvent> {
  11.     public StringEventHandler(Consumer<?> consumer) {
  12.         this.consumer = consumer;
  13.     }
  15.     // 外部可以傳入Consumer實現類，每處理一條訊息的時候，consumer的accept方法就會被執行一次
  16.     private Consumer<?> consumer;
  18.     @Override
  19.     public void onEvent(StringEvent event, long sequence, boolean endOfBatch) throws Exception {
  20.         log.info("sequence [{}], endOfBatch [{}], event : {}", sequence, endOfBatch, event);
  22.         // 這裡延時100ms，模擬消費事件的邏輯的耗時
  23.         Thread.sleep(100);
  25.         // 如果外部傳入了consumer，就要執行一次accept方法
  26.         if (null!=consumer) {
  27.             consumer.accept(null);
  28.         }
  29.     }
  30. }

事件生產者

  • 每當業務要生產一個事件時，就會呼叫事件生產者的onData方法，將業務資料作為入參傳進來，此時生產者會從環形佇列中取出一個事件例項（就是前面的事件工廠建立的），把業務資料傳給這個例項，再把例項正式釋出出去：
  1. package com.bolingcavalry.service;
  3. import com.lmax.disruptor.RingBuffer;
  5. public class StringEventProducer {
  6.     // 儲存資料的環形佇列
  7.     private final RingBuffer<StringEvent> ringBuffer;
  9.     public StringEventProducer(RingBuffer<StringEvent> ringBuffer) {
  10.         this.ringBuffer = ringBuffer;
  11.     }
  13.     public void onData(String content) {
  14.         // ringBuffer是個佇列，其next方法返回的是下最後一條記錄之後的位置，這是個可用位置
  15.         long sequence = ringBuffer.next();
  17.         try {
  18.             // sequence位置取出的事件是空事件
  19.             StringEvent stringEvent = ringBuffer.get(sequence);
  20.             // 空事件新增業務資訊
  21.             stringEvent.setValue(content);
  22.         } finally {
  23.             // 釋出
  24.             ringBuffer.publish(sequence);
  25.         }
  26.     }
  27. }

初始化邏輯

  • 開發一個spring bean，這裡面有disruptor的初始化邏輯，有幾處需要關注的地方稍後會說到：
  1. package com.bolingcavalry.service.impl;
  3. import com.bolingcavalry.service.*;
  4. import com.lmax.disruptor.RingBuffer;
  5. import com.lmax.disruptor.dsl.Disruptor;
  6. import com.lmax.disruptor.util.DaemonThreadFactory;
  7. import lombok.extern.slf4j.Slf4j;
  8. import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
  9. import org.springframework.stereotype.Service;
  10. import javax.annotation.PostConstruct;
  11. import java.time.LocalDateTime;
  12. import java.util.concurrent.Executor;
  13. import java.util.concurrent.Executors;
  14. import java.util.concurrent.ThreadFactory;
  15. import java.util.concurrent.atomic.AtomicInteger;
  16. import java.util.concurrent.atomic.AtomicLong;
  17. import java.util.function.Consumer;
  19. @Service
  20. @Slf4j
  21. public class BasicEventServiceImpl implements BasicEventService {
  23.     private static final int BUFFER_SIZE = 16;
  25.     private Disruptor<StringEvent> disruptor;
  27.     private StringEventProducer producer;
  29.     /**
  30.      * 統計訊息總數
  31.      */
  32.     private final AtomicLong eventCount = new AtomicLong();
  34.     @PostConstruct
  35.     private void init() {
  36.         Executor executor = Executors.newCachedThreadPool();
  38.         // 例項化
  39.         disruptor = new Disruptor<>(new StringEventFactory(),
  40.                 BUFFER_SIZE,
  41.                 new CustomizableThreadFactory("event-handler-"));
  43.         // 準備一個匿名類，傳給disruptor的事件處理類，
  44.         // 這樣每次處理事件時，都會將已經處理事件的總數打印出來
  45.         Consumer<?> eventCountPrinter = new Consumer<Object>() {
  46.             @Override
  47.             public void accept(Object o) {
  48.                 long count = eventCount.incrementAndGet();
  49.                 log.info("receive [{}] event", count);
  50.             }
  51.         };
  53.         // 指定處理類
  54.         disruptor.handleEventsWith(new StringEventHandler(eventCountPrinter));
  56.         // 啟動
  57.         disruptor.start();
  59.         // 生產者
  60.         producer = new StringEventProducer(disruptor.getRingBuffer());
  61.     }
  63.     @Override
  64.     public void publish(String value) {
  65.         producer.onData(value);
  66.     }
  68.     @Override
  69.     public long eventCount() {
  70.         return eventCount.get();
  71.     }
  72. }
  • 上述程式碼有以下幾點需要注意：
  1. publish方法給外部呼叫，用於釋出一個事件；
  2. eventCountPrinter是Consumer的實現類，被傳給了StringEventHandler，這樣StringEventHandler消費訊息的時候，eventCount就會增加，也就記下了已經處理的事件總數；
  3. Disruptor的構造方法中，BUFFER_SIZE表示環形佇列的大小，這裡故意設定為16，這樣可以輕易的將環形佇列填滿，此時再發布事件會不會導致環形佇列上的資料被覆蓋呢？稍後咱們可以測一下；
  4. 記得呼叫start方法；

web介面

再寫一個web介面類，這樣就可以通過瀏覽器驗證前面的程式碼了：

  1. package com.bolingcavalry.controller;
  3. import com.bolingcavalry.service.BasicEventService;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.boot.SpringApplication;
  6. import org.springframework.boot.autoconfigure.SpringBootApplication;
  7. import org.springframework.http.HttpStatus;
  8. import org.springframework.web.bind.annotation.*;
  10. import java.time.LocalDateTime;
  12. @RestController
  13. public class BasicEventController {
  15.     @Autowired
  16.     BasicEventService basicEventService;
  18.     @RequestMapping(value = "/{value}", method = RequestMethod.GET)
  19.     public String publish(@PathVariable("value") String value) {
  20.         basicEventService.publish(value);
  21.         return "success, " + LocalDateTime.now().toString();
  22.     }
  23. }

業務邏輯

  • 現在生產事件的介面已準備好，消費事件的程式碼也完成了，接下來就是如何呼叫生產事件的介面來驗證生產和消費是否正常，這裡我選擇使用單元測試來驗證；

  • 在disruptor-tutorials\basic-event\src\test\java目錄下新增測試類BasicEventServiceImplTest.java，測試邏輯是釋出了一百個事件，再驗證消費事件的數量是否也等於一百：

  1. package com.bolingcavalry.service.impl;
  3. import com.bolingcavalry.service.BasicEventService;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.junit.Test;
  6. import org.junit.runner.RunWith;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.test.context.SpringBootTest;
  9. import org.springframework.test.context.junit4.SpringRunner;
  10. import static org.junit.Assert.assertEquals;
  12. @RunWith(SpringRunner.class)
  13. @SpringBootTest
  14. @Slf4j
  15. public class BasicEventServiceImplTest {
  17.     @Autowired
  18.     BasicEventService basicEventService;
  20.     @Test
  21.     public void publish() throws InterruptedException {
  22.         log.info("start publich test");
  24.         int count = 100;
  26.         for(int i=0;i<count;i++) {
  27.             log.info("publich {}", i);
  28.             basicEventService.publish(String.valueOf(i));
  29.         }
  31.         // 非同步消費，因此需要延時等待
  32.         Thread.sleep(1000);
  33.         // 消費的事件總數應該等於釋出的事件數
  34.         assertEquals(count, basicEventService.eventCount());
  35.     }
  36. }
  • 編碼完成後，點選下圖紅框1中的按鈕執行單元測試，結果如紅框2所示，測試通過：

  • 聰明的您可能會產生疑惑：環形陣列大小隻有16，消費一個事件耗時很長(100毫秒)，那麼環形陣列中的事件還未消費完時如果還在釋出事件會發生什麼呢？新事件會覆蓋未消費的事件嗎？顯然不會，因為測試結果是通過的，那麼disruptor是怎麼做到的呢？其實從日誌上可以看出一些端倪，下圖是測試過程中日誌的末尾部分，紅框顯示，一直到測試快結束，釋出事件的執行緒還在執行釋出操作，這就意味著：如果消費速度過慢導致環形佇列裡放不進新的事件時，釋出事件的執行緒就會阻塞，知道環形佇列中可以放入事件為止：

  • 至此，disrupor的入門操作就完成了，咱們已經體驗過基本的釋出和消費功能，接下來的文章咱們會繼續深入學習其他更強大的功能；

你不孤單，欣宸原創一路相伴

  1. Java系列
  2. Spring系列
  3. Docker系列
  4. kubernetes系列
  5. 資料庫+中介軟體系列
  6. DevOps系列

歡迎關注公眾號：程式設計師欣宸

微信搜尋「程式設計師欣宸」，我是欣宸，期待與您一同暢遊Java世界...

https://github.com/zq2599/blog_demos