日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

分布式系統(tǒng)中必備的一個中間件就是消息隊列，通過消息隊列我們能對服務間進行異步解耦、流量消峰、實現(xiàn)最終一致性。

創(chuàng)新互聯(lián)公司-專業(yè)網(wǎng)站定制、快速模板網(wǎng)站建設、高性價比彰武網(wǎng)站開發(fā)、企業(yè)建站全套包干低至880元,成熟完善的模板庫,直接使用。一站式彰武網(wǎng)站制作公司更省心,省錢,快速模板網(wǎng)站建設找我們，業(yè)務覆蓋彰武地區(qū)。費用合理售后完善，十多年實體公司更值得信賴。

目前市面上已經(jīng)有 RabbitMQ、RochetMQ、ActiveMQ、Kafka等，有人會問：“Redis 適合做消息隊列么？”

在回答這個問題之前，我們先從本質思考：

• 消息隊列提供了什么特性？
• Redis 如何實現(xiàn)消息隊列？是否滿足存取需求？

今天，結合消息隊列的特點一步步帶大家分析使用 Redis 的 List 作為消息隊列的實現(xiàn)原理，并分享如何把 SpringBoot 與 Redission 整合運用到項目中。

什么是消息隊列

消息隊列是一種異步的服務間通信方式，適用于分布式和微服務架構。消息在被處理和刪除之前一直存儲在隊列上。

每條消息僅可被一位用戶處理一次。消息隊列可被用于分離重量級處理、緩沖或批處理工作以及緩解高峰期工作負載。

消息隊列

•  Producer：消息生產(chǎn)者，負責產(chǎn)生和發(fā)送消息到 Broker；
• Broker：消息處理中心。負責消息存儲、確認、重試等，一般其中會包含多個 queue；
• Consumer：消息消費者，負責從 Broker 中獲取消息，并進行相應處理；

     消息隊列的使用場景有哪些呢？

消息隊列在實際應用中包括如下四個場景：

• 應用耦合：發(fā)送方、接收方系統(tǒng)之間不需要了解雙方，只需要認識消息。多應用間通過消息隊列對同一消息進行處理，避免調(diào)用接口失敗導致整個過程失敗；
• 異步處理：多應用對消息隊列中同一消息進行處理，應用間并發(fā)處理消息，相比串行處理，減少處理時間；
• 限流削峰：廣泛應用于秒殺或搶購活動中，避免流量過大導致應用系統(tǒng)掛掉的情況；
• 消息驅動的系統(tǒng)：系統(tǒng)分為消息隊列、消息生產(chǎn)者、消息消費者，生產(chǎn)者負責產(chǎn)生消息，消費者(可能有多個)負責對消息進行處理；

消息隊列滿足哪些特性

消息有序性

消息是異步處理的，但是消費者需要按照生產(chǎn)者發(fā)送消息的順序來消費，避免出現(xiàn)后發(fā)送的消息被先處理的情況。

重復消息處理

生產(chǎn)者可能因為網(wǎng)絡問題出現(xiàn)消息重傳導致消費者可能會收到多條重復消息。

同樣的消息重復多次的話可能會造成一業(yè)務邏輯多次執(zhí)行，需要確保如何避免重復消費問題。

可靠性

一次保證消息的傳遞。如果發(fā)送消息時接收者不可用，消息隊列會保留消息，直到成功地傳遞它。

當消費者重啟后，可以繼續(xù)讀取消息進行處理，防止消息遺漏。

List 實現(xiàn)消息隊列

Redis 的列表（List）是一種線性的有序結構，可以按照元素被推入列表中的順序來存儲元素，能滿足「先進先出」的需求，這些元素既可以是文字數(shù)據(jù)，又可以是二進制數(shù)據(jù)。

LPUSH

生產(chǎn)者使用 LPUSH key element[element...] 將消息插入到隊列的頭部，如果 key 不存在則會創(chuàng)建一個空的隊列再插入消息。

如下，生產(chǎn)者向隊列 queue 先后插入了 「Java」「碼哥字節(jié)」「Go」，返回值表示消息插入隊列后的個數(shù)。

> LPUSH queue Java 碼哥字節(jié) Go
(integer) 3

RPOP

消費者使用 RPOP key 依次讀取隊列的消息，先進先出，所以 「Java」會先讀取消費：

> RPOP queue
"Java"
> RPOP queue
"碼哥字節(jié)"
> RPOP queue
"Go"

List隊列

實時消費問題

    65 哥：這么簡單就實現(xiàn)了么？

別高興的太早，LPUSH、RPOP 存在一個性能風險，生產(chǎn)者向隊列插入數(shù)據(jù)的時候，List 并不會主動通知消費者及時消費。

我們需要寫一個 while(true) 不停地調(diào)用 RPOP 指令，當有新消息就會返回消息，否則返回空。

程序需要不斷輪詢并判斷是否為空再執(zhí)行消費邏輯，這就會導致即使沒有新消息寫入到隊列，消費者也要不停地調(diào)用 RPOP 命令占用 CPU 資源。

    65 哥：要如何避免循環(huán)調(diào)用導致的 CPU 性能損耗呢？

Redis 提供了 BLPOP、BRPOP 阻塞讀取的命令，消費者在在讀取隊列沒有數(shù)據(jù)的時候自動阻塞，直到有新的消息寫入隊列，才會繼續(xù)讀取新消息執(zhí)行業(yè)務邏輯。

BRPOP queue 0

參數(shù) 0 表示阻塞等待時間無無限制

重復消費

• 消息隊列為每一條消息生成一個「全局 ID」；
• 生產(chǎn)者為每一條消息創(chuàng)建一條「全局 ID」，消費者把一件處理過的消息 ID 記錄下來判斷是否重復。

其實這就是冪等，對于同一條消息，消費者收到后處理一次的結果和多次的結果是一致的。

消息可靠性

     65 哥：消費者從 List 中讀取一條在消息處理過程中宕機了就會導致消息沒有處理完成，可是數(shù)據(jù)已經(jīng)沒有保存在 List 中了咋辦？

本質就是消費者在處理消息的時候崩潰了，就無法再還原消息，缺乏一個消息確認機制。

Redis 提供了 RPOPLPUSH、BRPOPLPUSH(阻塞)兩個指令，含義是從 List 從讀取消息的同時把這條消息復制到另一個 List 中（備份），并且是原子操作。

我們就可以在業(yè)務流程正確處理完成后再刪除隊列消息實現(xiàn)消息確認機制。如果在處理消息的時候宕機了，重啟后再從備份 List 中讀取消息處理。

LPUSH redisMQ 公眾號 碼哥字節(jié)
BRPOPLPUSH redisMQ redisMQBack

生產(chǎn)者用 LPUSH 把消息插入到 redisMQ 隊列中，消費者使用 BRPOPLPUSH 讀取消息「公眾號」，同時該消息會被插入到 「redisMQBack」隊列中。

如果消費成功則把「redisMQBack」的消息刪除即可，異常的話可以繼續(xù)從 「redisMQBack」再次讀取消息處理。

redis消息確認機制

需要注意的是，如果生產(chǎn)者消息發(fā)送的很快，而消費者處理速度慢就會導致消息堆積，給 Redis 的內(nèi)存帶來過大壓力。

Redission 實戰(zhàn)

在 Java 中，我們可以利用 Redission 封裝的 API 來快速實現(xiàn)隊列，接下來碼哥基于 SpringBoot 2.1.4 版本來交大家如何整合并實戰(zhàn)。

詳細 API 文檔大家可查閱：

??https://github.com/redisson/redisson/wiki/7.-Distributed-collections ??

添加依賴

  org.redisson
  redisson-spring-boot-starter
  3.16.7

添加 Redis 配置，碼哥的 Redis 沒有配置密碼，大家根據(jù)實際情況配置即可。

spring:
  application:
    name: redission
  redis:
    host: 127.0.0.1
    port: 6379
    ssl: false

Java 代碼實戰(zhàn)

RBlockingDeque 繼承 java.util.concurrent.BlockingDeque ，在使用過程中我們完全可以根據(jù)接口文檔來選擇合適的 API 去實現(xiàn)業(yè)務邏輯。

主要方法如下

碼哥采用了雙端隊列來舉例

@Slf4j
@Service
public class QueueService {
    @Autowired
    private RedissonClient redissonClient;
    private static final String REDIS_MQ = "redisMQ";
    /**
     * 發(fā)送消息到隊列頭部
     *
     * @param message
     */
    public void sendMessage(String message) {
        RBlockingDeque blockingDeque = redissonClient.getBlockingDeque(REDIS_MQ);
        try {
            blockingDeque.putFirst(message);
            log.info("將消息: {} 插入到隊列。", message);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    /**
     * 從隊列尾部阻塞讀取消息，若沒有消息，線程就會阻塞等待新消息插入，防止 CPU 空轉
     */
    public void onMessage() {
        RBlockingDeque blockingDeque = redissonClient.getBlockingDeque(REDIS_MQ);
        while (true) {
            try {
                String message = blockingDeque.takeLast();
                log.info("從隊列 {} 中讀取到消息：{}.", REDIS_MQ, message);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

單元測試

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RedissionApplication.class)
public class RedissionApplicationTests {
    @Autowired
    private QueueService queueService;
    @Test
    public void testQueue() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                queueService.sendMessage("消息" + i);
            }
        }).start();
        new Thread(() -> queueService.onMessage()).start();
        Thread.currentThread().join();
    }
}

總結

可以使用 List 數(shù)據(jù)結構來實現(xiàn)消息隊列，滿足先進先出。為了實現(xiàn)消息可靠性，Redis 提供了 BRPOPLPUSH 命令是解決。

Redis 是一個非常輕量級的鍵值數(shù)據(jù)庫，部署一個 Redis 實例就是啟動一個進程，部署 Redis 集群，也就是部署多個 Redis 實例。

而 Kafka、RabbitMQ 部署時，涉及額外的組件，例如 Kafka 的運行就需要再部署 ZooKeeper。相比 Redis 來說，Kafka 和 RabbitMQ 一般被認為是重量級的消息隊列。

需要注意的是，我們要避免生產(chǎn)者過快，消費者過慢導致的消息堆積占用 Redis 的內(nèi)存。

在消息量不大的情況下使用 Redis 作為消息隊列，他能給我們帶來高性能的消息讀寫，這似乎也是一個很好消息隊列解決方案。

大家覺得是否合適作為消息隊列呢？點贊讓我看看吧

文章名稱：Redis 使用 List 實現(xiàn)消息隊列的利與弊
文章轉載：http://www.5511xx.com/article/cccpsce.html