代码拉取完成,页面将自动刷新
同步操作将从 SnailClimb/JavaGuide 强制同步,此操作会覆盖自 Fork 仓库以来所做的任何修改,且无法恢复!!!
确定后同步将在后台操作,完成时将刷新页面,请耐心等待。
本文来自读者 PR。
一个消息中心,最基本的需要支持多生产者、多消费者,例如下:
class Scratch {
public static void main(String[] args) {
// 实际中会有 nameserver 服务来找到 broker 具体位置以及 broker 主从信息
Broker broker = new Broker();
Producer producer1 = new Producer();
producer1.connectBroker(broker);
Producer producer2 = new Producer();
producer2.connectBroker(broker);
Consumer consumer1 = new Consumer();
consumer1.connectBroker(broker);
Consumer consumer2 = new Consumer();
consumer2.connectBroker(broker);
for (int i = 0; i < 2; i++) {
producer1.asyncSendMsg("producer1 send msg" + i);
producer2.asyncSendMsg("producer2 send msg" + i);
}
System.out.println("broker has msg:" + broker.getAllMagByDisk());
for (int i = 0; i < 1; i++) {
System.out.println("consumer1 consume msg:" + consumer1.syncPullMsg());
}
for (int i = 0; i < 3; i++) {
System.out.println("consumer2 consume msg:" + consumer2.syncPullMsg());
}
}
}
class Producer {
private Broker broker;
public void connectBroker(Broker broker) {
this.broker = broker;
}
public void asyncSendMsg(String msg) {
if (broker == null) {
throw new RuntimeException("please connect broker first");
}
new Thread(() -> {
broker.sendMsg(msg);
}).start();
}
}
class Consumer {
private Broker broker;
public void connectBroker(Broker broker) {
this.broker = broker;
}
public String syncPullMsg() {
return broker.getMsg();
}
}
class Broker {
// 对应 RocketMQ 中 MessageQueue,默认情况下 1 个 Topic 包含 4 个 MessageQueue
private LinkedBlockingQueue<String> messageQueue = new LinkedBlockingQueue(Integer.MAX_VALUE);
// 实际发送消息到 broker 服务器使用 Netty 发送
public void sendMsg(String msg) {
try {
messageQueue.put(msg);
// 实际会同步或异步落盘,异步落盘使用的定时任务定时扫描落盘
} catch (InterruptedException e) {
}
}
public String getMsg() {
try {
return messageQueue.take();
} catch (InterruptedException e) {
}
return null;
}
public String getAllMagByDisk() {
StringBuilder sb = new StringBuilder("\n");
messageQueue.iterator().forEachRemaining((msg) -> {
sb.append(msg + "\n");
});
return sb.toString();
}
}
问题:
- 引入锁来实现串行
- 前一个消费阻塞时后面都会被阻塞
注意,就算是事务消息最后回滚了也不会物理删除,只会逻辑删除该消息
加分项咯
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。