原文連結：Pulsar の 保證訊息的順序性、冪等性和可靠性
一、背景
前面兩篇文章，已經介紹了關於Pulsar消費者的詳細使用和自研的Pulsar元件。
接下來，將簡單分析如何保證訊息的順序性、冪等性和可靠性；但並不會每個分析都會進行程式碼實戰，進行程式碼實戰的都是比較有意思的點，如消費訊息如何保證順序性和冪等性，而其他的其實都是比較簡單的，就不做程式碼實戰了。
二、特性分析
2.1、順序性
保證訊息是按順序傳送，按順序消費，一個接著一個。
2.1.1、活動圖
2.1.2、分析
producer：
傳送者保證訊息的順序性其實是比較簡單的：
- 利用單佇列傳送
- 一個業務對應一個佇列
- 一個佇列只能由一個消費者監聽消費
- 利用 Pulsar 的分割槽Topic
- producer傳送訊息時需要指定key屬性，Pulsar自動會根據Key值將訊息分配到指定的分割槽中
- 支援多個消費者消費，多個消費者可以監聽同一個分割槽，但是相同的Key只會分配給同一個消費者
生產者這裡就不做什麼實戰的，都是比較簡單的點，沒啥好說的。
consumer：
消費者保證訊息的順序性有下面兩種方式：
- 當前執行緒執行
- 單執行緒執行保證了消費的順序性
- 消費效率低
- 自定義執行緒池列表非同步併發消費
- 如果直接使用執行緒池，那麼雖然能提高消費效率，但是並不能保證順序性
- 這裡我們會自定義執行緒池列表，列表中的執行緒池的核心執行緒數和最大執行緒數都是1，保證順序消費
- Producer傳送的訊息體中，需指定key，我們會根據key#hashCode定位到對應的執行緒池，這裡參考HashMap的做法。
2.1.3、程式碼實戰
消費者保證訊息順序性的第二點的實現還是比較有意思的：如何自定義執行緒池列表、如何根據訊息的key來定位執行緒池。
程式碼如下：
- 傳送訊息：
/**
* 指定key傳送訊息
* @author winfun
**/
@Slf4j
public class ThirdProducerDemo {
public static void main(String[] args) throws PulsarClientException {
PulsarClient client = PulsarClient.builder()
.serviceUrl("pulsar://127.0.0.1:6650")
.build();
ProducerBuilder<String> productBuilder = client.newProducer(Schema.STRING).topic("winfun/study/test-topic3")
.blockIfQueueFull(Boolean.TRUE).batchingMaxMessages(100).enableBatching(Boolean.TRUE).sendTimeout(3, TimeUnit.SECONDS);
Producer<String> producer = productBuilder.create();
for (int i = 0; i < 100; i++) {
MsgDTO msgDTO = new MsgDTO();
String content = "hello"+i;
String key;
if (content.contains("1")){
key = "k213e434y1df";
}else if (content.contains("2")){
key = "keasdgashgfy2";
}else {
key = "other";
}
msgDTO.setId(key);
msgDTO.setContent(content);
producer.send(JSONUtil.toJsonStr(msgDTO));
}
producer.close();
}
}
- 消費訊息
/**
* 順序性消費-消費者demo
* @author: winfun
**/
@Slf4j
@PulsarListener(topics = {"test-topic3"})
public class SuccessionConsumerListener extends BaseMessageListener {
List<ExecutorService> executorServiceList = new ArrayList<>();
/**
* 初始化自定義執行緒池列表
*/
@PostConstruct
public void initCustomThreadPool(){
for (int i = 0; i < 10; i++) {
/**
* 1、核心執行緒數和最大執行緒數都為1，避免多執行緒消費導致順序被打亂
* 2、使用有界佇列，設定最大長度，避免無限任務數導致OOM
* 3、使用CallerRunsPolicy拒絕策略，讓當前執行緒執行，避免訊息丟失，也可以直接讓消費者執行當前任務，阻塞住其他任務，也能保證順序性
*/
ExecutorService threadPoolExecutor = new ThreadPoolExecutor(
1,
1,
60,
TimeUnit.MINUTES,
new LinkedBlockingDeque<>(100),
new ThreadFactoryBuilder().setNameFormat(String.format("custom-thread-pool-%d",i)).get(),
new ThreadPoolExecutor.CallerRunsPolicy()
);
this.executorServiceList.add(threadPoolExecutor);
}
}
/**
* 消費訊息
* 自定義監聽器實現方法
* 訊息如何響應由開發者決定：
* Consumer#acknowledge
* Consumer#reconsumeLater
* Consumer#negativeAcknowledge
*
* @param consumer 消費者
* @param msg 訊息
*/
@Override
protected void doReceived(Consumer<String> consumer, Message<String> msg) {
String value = msg.getValue();
MsgDTO msgDTO = JSONUtil.toBean(value, MsgDTO.class);
// 匹配列表中對應的執行緒池
int index = (this.executorServiceList.size()-1)&this.spreed(msgDTO.getId().hashCode());
log.info("成功獲取執行緒池列表索引，msgId is {}, index is {}",msgDTO.getId(),index);
ExecutorService executorService = this.executorServiceList.get(index);
executorService.execute(()->{
log.info("成功消費訊息，threadName is {},msg is {}",Thread.currentThread().getName(),msg);
consumer.acknowledgeAsync(msg);
});
}
/**
* hashCode擴充套件，保證hashCode前後十六位都能完美進行位運算
* @param hashCode
* @return
*/
private int spreed(int hashCode){
return (hashCode ^ (hashCode >>> 16)) & hashCode;
}
/***
* 是否開啟非同步消費，預設開啟
* @return {@link Boolean }
**/
@Override
public Boolean enableAsync() {
// 首先關閉執行緒池非同步併發消費
return Boolean.FALSE;
}
}
2.2、冪等性
冪等性的話，我們主要是分析一下消費者的，如何保證消費者只正確消費一次訊息還是非常重要的。
2.2.1、活動圖
2.2.2、分析
producer：
生產者如何保證冪等性，感覺這個話題沒什麼好討論的，如果發生失敗就重新發送，否則就正常傳送就好了。
consumer：
消費者保證訊息冪等性，最主要是利用中間表來儲存消費記錄：
- 本地新增表來儲存訊息消費記錄
- 在訊息消費前，先判斷MessageId判斷是否存在消費記錄
- 如果存在，直接響應
- 如果不存在，則開啟本地事務，接著進行訊息消費
- 當成功消費時提交事務，否則回滾
2.2.3、程式碼實戰
如何利用消費記錄表和本地事務來完成訊息消費的冪等性，看下面程式碼：
- 傳送訊息
**
*
* @author winfun
**/
@Slf4j
public class FourthProducerDemo {
public static void main(String[] args) throws PulsarClientException {
PulsarClient client = PulsarClient.builder()
.serviceUrl("pulsar://127.0.0.1:6650")
.build();
ProducerBuilder<String> productBuilder = client.newProducer(Schema.STRING).topic("winfun/study/test-topic4")
.blockIfQueueFull(Boolean.TRUE).batchingMaxMessages(100).enableBatching(Boolean.TRUE).sendTimeout(3, TimeUnit.SECONDS);
Producer<String> producer = productBuilder.create();
for (int i = 0; i < 20; i++) {
MsgDTO msgDTO = new MsgDTO();
String content = "hello"+i;
String key;
if (content.contains("1")){
key = "k213e434y1df";
}else if (content.contains("2")){
key = "keasdgashgfy2";
}else {
key = "other";
}
msgDTO.setId(key);
msgDTO.setContent(content);
producer.send(JSONUtil.toJsonStr(msgDTO));
}
producer.close();
}
}
- 消費訊息
package com.github.howinfun.consumer.idempotent;
import cn.hutool.json.JSONUtil;
import com.github.howinfun.core.entity.MessageConsumeRecord;
import com.github.howinfun.core.service.MessageConsumeRecordService;
import com.github.howinfun.dto.MsgDTO;
import io.github.howinfun.listener.BaseMessageListener;
import io.github.howinfun.listener.PulsarListener;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import jodd.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
/**
* 冪等性消費-消費者demo
* @author: winfun
* @date: 2021/9/2 12:49 下午
**/
@Slf4j
@PulsarListener(topics = {"test-topic4"})
public class IdempotentConsumerListener extends BaseMessageListener {
List<ExecutorService> executorServiceList = new ArrayList<>();
@Autowired
private MessageConsumeRecordService service;
/**
* 初始化自定義執行緒池列表
*/
@PostConstruct
public void initCustomThreadPool(){
for (int i = 0; i < 10; i++) {
/**
* 1、核心執行緒數和最大執行緒數都為1，避免多執行緒消費導致順序被打亂
* 2、使用有界佇列，設定最大長度，避免無限任務數導致OOM
* 3、使用CallerRunsPolicy拒絕策略，讓當前執行緒執行，避免訊息丟失，也可以直接讓消費者執行當前任務，阻塞住其他任務，也能保證順序性
*/
ExecutorService threadPoolExecutor = new ThreadPoolExecutor(
1,
1,
60,
TimeUnit.MINUTES,
new LinkedBlockingDeque<>(100),
new ThreadFactoryBuilder().setNameFormat(String.format("custom-thread-pool-%d",i)).get(),
new ThreadPoolExecutor.CallerRunsPolicy()
);
this.executorServiceList.add(threadPoolExecutor);
}
}
/**
* 消費訊息
* 自定義監聽器實現方法
* 訊息如何響應由開發者決定：
* Consumer#acknowledge
* Consumer#reconsumeLater
* Consumer#negativeAcknowledge
*
* @param consumer 消費者
* @param msg 訊息
*/
@Override
protected void doReceived(Consumer<String> consumer, Message<String> msg) {
boolean flag = preReceived(msg);
if (Boolean.FALSE.equals(flag)){
String value = msg.getValue();
MsgDTO msgDTO = JSONUtil.toBean(value, MsgDTO.class);
int index = (this.executorServiceList.size()-1)&this.spreed(msgDTO.getId().hashCode());
log.info("成功獲取執行緒池列表索引，msgId is {}, index is {}",msgDTO.getId(),index);
ExecutorService executorService = this.executorServiceList.get(index);
executorService.execute(()->{
try {
this.doInnerReceived(consumer,msg);
} catch (PulsarClientException e) {
log.error("訊息消費失敗",e);
}
});
}else {
log.info("此訊息的消費記錄已存在，直接響應,messageId is {}", msg.getMessageId().toString());
try {
consumer.acknowledge(msg);
} catch (PulsarClientException e) {
log.error("訊息提交失敗",e);
}
}
}
/**
* 消費前判斷，messageId是否存在對應的消費記錄
* @param msg 訊息
* @return 存在結果
*/
private boolean preReceived(Message<String> msg){
MessageConsumeRecord record = this.service.getByMessageId(msg.getMessageId().toString());
if (Objects.isNull(record)){
return false;
}
return true;
}
/**
* 訊息消費
* @param consumer 消費者
* @param msg 訊息
*/
@Transactional(rollbackFor = Exception.class)
public void doInnerReceived(Consumer<String> consumer,Message<String> msg) throws PulsarClientException {
String messageContent = msg.getValue();
String messageId = msg.getMessageId().toString();
log.info("成功消費訊息，threadName is {},msg is {}",Thread.currentThread().getName(),messageContent);
this.service.save(new MessageConsumeRecord()
.setMessageId(messageId)
.setMessageContent(messageContent)
.setCreateTime(new Date()));
// 模擬重複消費，如果訊息內容包含8，則插入資料庫，但是不響應
if (messageContent.contains("8")){
log.info("訊息已被消費入庫，但不響應，模擬重複消費，messageId is {},messageContent is {}",messageId,messageContent);
}else {
consumer.acknowledge(msg);
}
}
/**
* hashCode擴充套件，保證hashCode前後十六位都能完美計算
* @param hashCode
* @return
*/
private int spreed(int hashCode){
return (hashCode ^ (hashCode >>> 16)) & hashCode;
}
/***
* 是否開啟非同步消費，預設開啟
* @return {@link Boolean }
**/
@Override
public Boolean enableAsync() {
// 首先關閉執行緒池非同步併發消費
return Boolean.FALSE;
}
}
2.3、可靠性
2.3.1、活動圖
生產者：
消費者：
關於保證訊息的可靠性，我們只分析 Producer 和 Consuemr，Pulsar伺服器就不分析了。
2.3.2、分析
producer：
生產者主要還是利用中間表來保證訊息傳送的可靠性：
- 傳送訊息前，先插入一條傳送記錄表
- 接著開啟本地事務，開始傳送訊息
- 傳送完畢，接到broker返回的響應
- 更新發送記錄為已傳送
- 開啟定時任務，定時掃描未傳送的記錄，重新進行傳送
consumer：
消費者保證訊息的可靠性，只需要利用Pulsar提供的重試策略即可：
- 開啟重試策略，指定重試次數、重試佇列和死信佇列
- 捕獲異常，呼叫reconsumeLater方法進行重新消費
- 監控死信佇列，即使進行訊息消費異常人工處理