當前位置：首頁 >

Java多线程消费消息

發布時間：2023/11/21 57 coder

生活随笔收集整理的這篇文章主要介紹了 Java多线程消费消息小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

多線程消費消息

關鍵詞：Java，多線程，消息隊列，rocketmq

多線程一個用例之一就是消息的快速消費，比如我們有一個消息隊列我們希望以更快的速度消費消息，假如我們用的是rocketmq，我們從中獲取消息，然后使用多線程處理。

代碼地址Github

實現思路

不停的拉取消息
將拉取的消息分片
多個線程一起消費每一片消息
將所有消息消費完成后，接著拉取新的消息

代碼

CrazyTask

這是一個抽象類，針對不同的任務可能有不同的處理邏輯，對于不同的任務去繼承這個CrazyTask 實現他的process方法。

package crazyConsumer;

import com.google.common.collect.Lists;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

/**
 * {@code @author:} keboom
 * {@code @date:} 2023/11/17
 * {@code @description:}
 */
public abstract class CrazyTask {
    String taskName;
    int threadNum;
    volatile boolean isTerminated;
    // every partition data num.
    // for example: I receive 5 messages, partitionDataNum is 2, then i will partition 5 messages to 3 parts, 2,2,1
    int partitionDataCount = 2;

    abstract void process(Message message);

    void doExecute(SimpleConsumer consumer) {
        while (true) {
            // 此消費者每次主動拉取消息隊列中消息
            List<Message> messages = consumer.receive();
            if (messages.isEmpty()) {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                continue;
            }
            // 獲取處理此topic或者說處理此類型task的線程池
            ExecutorService executor = CrazyTaskUtil.getOrInitExecutor(taskName, threadNum);
            // 將消息分片，每個線程處理一部分消息
            List<List<Message>> partition = Lists.partition(messages, partitionDataCount);
            // 以消息分片數初始化CountDownLatch，每個線程處理完一片消息，countDown一次
            // 當countDownLatch為0時，說明所有消息都處理完了，countDownLatch.await();繼續向下執行
            CountDownLatch countDownLatch = new CountDownLatch(partition.size());

            partition.forEach(messageList -> {
                executor.execute(() -> {
                    messageList.forEach(message -> {
                        process(message);
                        consumer.ack(message);
                    });
                    countDownLatch.countDown();
                });
            });
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            if (isTerminated) {
                break;
            }
        }
        // 釋放線程池
        CrazyTaskUtil.shutdownThreadPool(taskName);
    }

    void terminate() {
        isTerminated = true;
        System.out.println();
        System.out.println(taskName + " shut down");
    }

    public String getTaskName() {
        return taskName;
    }
}

PhoneTask

package crazyConsumer;

/**
 * {@code @author:} keboom
 * {@code @date:} 2023/11/17
 * {@code @description:}
 */
public class PhoneTask extends CrazyTask {

    public PhoneTask(String taskName, int threadNum) {
        this.taskName = taskName;
        // default thread num
        this.threadNum = threadNum;
        this.isTerminated = false;
    }

    @Override
    void process(Message message) {
        System.out.println(Thread.currentThread().getName() +"  process  "+ message.toString());
        try {
            Thread.sleep(30);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public String toString() {
        return "PhoneTask{" +
                "taskName='" + taskName + '\'' +
                ", threadNum=" + threadNum +
                ", isTerminated=" + isTerminated +
                '}';
    }
}

EmailTask

package crazyConsumer;

/**
 * {@code @author:} keboom
 * {@code @date:} 2023/11/17
 * {@code @description:}
 */
public class EmailTask extends CrazyTask{

    public EmailTask(String taskName, int threadNum) {
        this.taskName = taskName;
        // default thread num
        this.threadNum = threadNum;
        this.isTerminated = false;
    }

    @Override
    void process(Message message) {
        // do something
        System.out.println(Thread.currentThread().getName() +"  process  "+ message.toString());
        try {
            Thread.sleep(20);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public String toString() {
        return "EmailTask{" +
                "taskName='" + taskName + '\'' +
                ", threadNum=" + threadNum +
                ", isTerminated=" + isTerminated +
                '}';
    }
}

CrazyTaskUtil

創建銷毀線程池的工具類

package crazyConsumer;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.Map;
import java.util.concurrent.*;

/**
 * {@code @author:} keboom
 * {@code @date:} 2023/11/17
 * {@code @description:}
 */
public class CrazyTaskUtil {

    private static final Map<String, ExecutorService> executors = new ConcurrentHashMap<>();

    public static ExecutorService getOrInitExecutor(String taskName, int threadNum) {
        ExecutorService executorService = executors.get(taskName);
        if (executorService == null) {
            synchronized (CrazyTaskUtil.class) {
                executorService = executors.get(taskName);
                if (executorService == null) {
                    executorService = initPool(taskName, threadNum);
                    executors.put(taskName, executorService);
                }
            }
        }
        return executorService;
    }

    private static ExecutorService initPool(String taskName, int threadNum) {
        // init pool
        return new ThreadPoolExecutor(threadNum, threadNum,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new ThreadFactoryBuilder().setNameFormat(taskName + "-%d").build());
    }

    public static void shutdownThreadPool(String taskName) {
        ExecutorService remove = executors.remove(taskName);
        if (remove != null) {
            remove.shutdown();
        }
    }

}

Main

程序入口

package crazyConsumer;

import java.util.ArrayList;

/**
 * {@code @author:} keboom
 * {@code @date:} 2023/11/17
 * {@code @description:}
 */
public class Main {

    /**
     * 一種多線程消費場景。比如我們有一個消費隊列，里面有各種消息，我們需要盡快的消費他們，不同的消息對應不同的業務
     *
     * @param args
     */
    public static void main(String[] args) throws InterruptedException {

        // 比方說我們這個有rocketmq不同主題的consumer
        /*
        List<MessageView> messageViewList = null;
        try {
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                //消費處理完成后，需要主動調用ACK提交消費結果。
                try {
                    simpleConsumer.ack(messageView);
                } catch (ClientException e) {
                    e.printStackTrace();
                }
            });
        } catch (ClientException e) {
            //如果遇到系統流控等原因造成拉取失敗，需要重新發起獲取消息請求。
            e.printStackTrace();
        }

         */

        // 想要實現多線程消費消息，我們希望有一個任務，此任務能夠不停的拉取消息，然后創建子線程池去消費消息。
        // 停止任務后，需要將任務中的消息消費完后，再關閉任務。

        ArrayList<CrazyTask> tasks = new ArrayList<>();
        tasks.add(new PhoneTask("phoneTask", 2));
        tasks.add(new EmailTask("emailTask", 3));

        for (CrazyTask task : tasks) {
            new Thread(() -> {
                task.doExecute(new SimpleConsumer("topic"+task.getTaskName().charAt(0), "tagA"));
            }).start();
        }

        // task running
        Thread.sleep(150);

        // task terminated
        tasks.forEach(CrazyTask::terminate);
    }
}

最終執行結果

receive message: [Message{messageBody='topice-tagA-0-1700470193487'}, Message{messageBody='topice-tagA-1-1700470193487'}, Message{messageBody='topice-tagA-2-1700470193487'}, Message{messageBody='topice-tagA-3-1700470193487'}, Message{messageBody='topice-tagA-4-1700470193487'}]
receive message: [Message{messageBody='topicp-tagA-0-1700470193487'}, Message{messageBody='topicp-tagA-1-1700470193487'}, Message{messageBody='topicp-tagA-2-1700470193487'}, Message{messageBody='topicp-tagA-3-1700470193487'}, Message{messageBody='topicp-tagA-4-1700470193487'}]
phoneTask-0  process  Message{messageBody='topicp-tagA-0-1700470193487'}
emailTask-1  process  Message{messageBody='topice-tagA-2-1700470193487'}
emailTask-0  process  Message{messageBody='topice-tagA-0-1700470193487'}
phoneTask-1  process  Message{messageBody='topicp-tagA-2-1700470193487'}
emailTask-2  process  Message{messageBody='topice-tagA-4-1700470193487'}
ack message: Message{messageBody='topice-tagA-2-1700470193487'}
emailTask-1  process  Message{messageBody='topice-tagA-3-1700470193487'}
ack message: Message{messageBody='topice-tagA-4-1700470193487'}
ack message: Message{messageBody='topice-tagA-0-1700470193487'}
emailTask-0  process  Message{messageBody='topice-tagA-1-1700470193487'}
ack message: Message{messageBody='topicp-tagA-2-1700470193487'}
ack message: Message{messageBody='topicp-tagA-0-1700470193487'}
phoneTask-0  process  Message{messageBody='topicp-tagA-1-1700470193487'}
phoneTask-1  process  Message{messageBody='topicp-tagA-3-1700470193487'}
ack message: Message{messageBody='topice-tagA-1-1700470193487'}
ack message: Message{messageBody='topice-tagA-3-1700470193487'}
receive message: [Message{messageBody='topice-tagA-0-1700470193570'}, Message{messageBody='topice-tagA-1-1700470193570'}, Message{messageBody='topice-tagA-2-1700470193570'}, Message{messageBody='topice-tagA-3-1700470193570'}, Message{messageBody='topice-tagA-4-1700470193570'}]
emailTask-0  process  Message{messageBody='topice-tagA-2-1700470193570'}
emailTask-2  process  Message{messageBody='topice-tagA-0-1700470193570'}
emailTask-1  process  Message{messageBody='topice-tagA-4-1700470193570'}
ack message: Message{messageBody='topicp-tagA-3-1700470193487'}
ack message: Message{messageBody='topicp-tagA-1-1700470193487'}
phoneTask-1  process  Message{messageBody='topicp-tagA-4-1700470193487'}
ack message: Message{messageBody='topice-tagA-0-1700470193570'}
ack message: Message{messageBody='topice-tagA-4-1700470193570'}
ack message: Message{messageBody='topice-tagA-2-1700470193570'}
emailTask-0  process  Message{messageBody='topice-tagA-3-1700470193570'}
emailTask-2  process  Message{messageBody='topice-tagA-1-1700470193570'}
ack message: Message{messageBody='topicp-tagA-4-1700470193487'}
receive message: [Message{messageBody='topicp-tagA-0-1700470193618'}, Message{messageBody='topicp-tagA-1-1700470193618'}, Message{messageBody='topicp-tagA-2-1700470193618'}, Message{messageBody='topicp-tagA-3-1700470193618'}, Message{messageBody='topicp-tagA-4-1700470193618'}]
phoneTask-0  process  Message{messageBody='topicp-tagA-0-1700470193618'}
phoneTask-1  process  Message{messageBody='topicp-tagA-2-1700470193618'}
ack message: Message{messageBody='topice-tagA-1-1700470193570'}
ack message: Message{messageBody='topice-tagA-3-1700470193570'}
receive message: [Message{messageBody='topice-tagA-0-1700470193634'}, Message{messageBody='topice-tagA-1-1700470193634'}, Message{messageBody='topice-tagA-2-1700470193634'}, Message{messageBody='topice-tagA-3-1700470193634'}, Message{messageBody='topice-tagA-4-1700470193634'}]
emailTask-1  process  Message{messageBody='topice-tagA-0-1700470193634'}
emailTask-0  process  Message{messageBody='topice-tagA-4-1700470193634'}
emailTask-2  process  Message{messageBody='topice-tagA-2-1700470193634'}
ack message: Message{messageBody='topicp-tagA-2-1700470193618'}
ack message: Message{messageBody='topicp-tagA-0-1700470193618'}
phoneTask-0  process  Message{messageBody='topicp-tagA-1-1700470193618'}
phoneTask-1  process  Message{messageBody='topicp-tagA-3-1700470193618'}

phoneTask shut down

emailTask shut down
ack message: Message{messageBody='topice-tagA-0-1700470193634'}
ack message: Message{messageBody='topice-tagA-2-1700470193634'}
emailTask-1  process  Message{messageBody='topice-tagA-1-1700470193634'}
ack message: Message{messageBody='topice-tagA-4-1700470193634'}
emailTask-2  process  Message{messageBody='topice-tagA-3-1700470193634'}
ack message: Message{messageBody='topicp-tagA-3-1700470193618'}
ack message: Message{messageBody='topicp-tagA-1-1700470193618'}
phoneTask-1  process  Message{messageBody='topicp-tagA-4-1700470193618'}
ack message: Message{messageBody='topice-tagA-3-1700470193634'}
ack message: Message{messageBody='topice-tagA-1-1700470193634'}
ack message: Message{messageBody='topicp-tagA-4-1700470193618'}

可以看到結果是，當每次收到的消息消費完后會拉取新的消息。當執行shutdown任務時，會將當前任務執行完后再銷毀線程池。

總結

以上是生活随笔為你收集整理的Java多线程消费消息的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： AtCoder Beginner Con
下一篇： Java Stream中的API你都用过