Java多线程消费消息
生活随笔
收集整理的這篇文章主要介紹了
Java多线程消费消息
小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.
多線程消費(fèi)消息
關(guān)鍵詞：Java，多線程，消息隊(duì)列，rocketmq
多線程一個(gè)用例之一就是消息的快速消費(fèi)，比如我們有一個(gè)消息隊(duì)列我們希望以更快的速度消費(fèi)消息，假如我們用的是rocketmq，我們從中獲取消息，然后使用多線程處理。
代碼地址Github
實(shí)現(xiàn)思路
- 不停的拉取消息
- 將拉取的消息分片
- 多個(gè)線程一起消費(fèi)每一片消息
- 將所有消息消費(fèi)完成后，接著拉取新的消息
代碼
CrazyTask
這是一個(gè)抽象類，針對(duì)不同的任務(wù)可能有不同的處理邏輯，對(duì)于不同的任務(wù)去繼承這個(gè)CrazyTask 實(shí)現(xiàn)他的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) {
// 此消費(fèi)者每次主動(dòng)拉取消息隊(duì)列中消息
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);
// 將消息分片，每個(gè)線程處理一部分消息
List<List<Message>> partition = Lists.partition(messages, partitionDataCount);
// 以消息分片數(shù)初始化CountDownLatch，每個(gè)線程處理完一片消息，countDown一次
// 當(dāng)countDownLatch為0時(shí)，說明所有消息都處理完了，countDownLatch.await();繼續(xù)向下執(zhí)行
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
創(chuàng)建銷毀線程池的工具類
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 {
/**
* 一種多線程消費(fèi)場(chǎng)景。比如我們有一個(gè)消費(fèi)隊(duì)列，里面有各種消息，我們需要盡快的消費(fèi)他們，不同的消息對(duì)應(yīng)不同的業(yè)務(wù)
*
* @param args
*/
public static void main(String[] args) throws InterruptedException {
// 比方說我們這個(gè)有rocketmq不同主題的consumer
/*
List<MessageView> messageViewList = null;
try {
messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
messageViewList.forEach(messageView -> {
System.out.println(messageView);
//消費(fèi)處理完成后，需要主動(dòng)調(diào)用ACK提交消費(fèi)結(jié)果。
try {
simpleConsumer.ack(messageView);
} catch (ClientException e) {
e.printStackTrace();
}
});
} catch (ClientException e) {
//如果遇到系統(tǒng)流控等原因造成拉取失敗，需要重新發(fā)起獲取消息請(qǐng)求。
e.printStackTrace();
}
*/
// 想要實(shí)現(xiàn)多線程消費(fèi)消息，我們希望有一個(gè)任務(wù)，此任務(wù)能夠不停的拉取消息，然后創(chuàng)建子線程池去消費(fèi)消息。
// 停止任務(wù)后，需要將任務(wù)中的消息消費(fèi)完后，再關(guān)閉任務(wù)。
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);
}
}
最終執(zhí)行結(jié)果
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'}
可以看到結(jié)果是，當(dāng)每次收到的消息消費(fèi)完后會(huì)拉取新的消息。當(dāng)執(zhí)行shutdown任務(wù)時(shí)，會(huì)將當(dāng)前任務(wù)執(zhí)行完后再銷毀線程池。
總結(jié)
以上是生活随笔為你收集整理的Java多线程消费消息的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： AtCoder Beginner Con
- 下一篇： Java Stream中的API你都用过