RocketMQHA机制
2020-12-22 本文已影响0人
93张先生
概览
RocketMQ作为一款优秀的分布式消息中间件，分布式系统的一个很重要的特点就是要保证系统的高可用(HA)，RocketMQ则是通过主从同步机制保证系统的高可用。
下面是两主两从的主从同步原理图。
image.png
主从两节点优点
- 数据备份：保证了两/多台机器上的数据冗余，特别是在主从同步复制的情况下，一定程度上保证了Master出现不可恢复的故障以后，数据不丢失。
- 高可用性：即使Master掉线， Consumer会自动重连到对应的Slave机器，不会出现消费停滞的情况。
- 提高性能：主要表现为可分担Master读的压力，当从Master拉取消息，拉取消息的最大物理偏移与本地存储的最大物理偏移的差值超过一定值，会转向Slave(默认brokerId=1)进行读取，减轻了Master压力。
- 消费实时：master宕机后消费者可以从slave上消费保证消息的实时性，但是slave不能接收producer发送的消息，slave只能同步master数据（RocketMQ4.5版本之前），4.5版本开始增加多副本机制，根据RAFT算法，master宕机会自动选择其中一个副本节点作为master保证消息可以正常的生产消费。
主从数据同步有两种方式同步复制、异步复制
| 复制方式 | 优点 | 缺点 | 适应场景 |
|---|---|---|---|
| 同步复制 | slave保证了与master一致的数据副本，如果master宕机，数据依然在slave中找到其数据和master的数据一致 | 由于需要slave确认效率上会有一定的损失 | 数据可靠性要求很高的场景 |
| 异步复制 | 无需等待slave确认消息是否存储成功效率上要高于同步复制 | 如果master宕机，由于数据同步有延迟导致slave和master存在一定程度的数据不一致问题 | 数据可靠性要求一般的场景 |
CommitLog复制
主从节点同步只复制commitlog消息信息，consumequeue、indexfile所索引文件不会同步，会由从节点Broker的commitlog文件重新生成本机的consumequeue、indexfile的索引信息。
-
HAService：Master和Slave通信的服务类；包含Slave作为客户端的HAClient类对象。
-
HAClient：Slave向Master建立连接，发送Offset数据请求，并处理Master返回请求的类对象；HAClient作为Slave向Master通信的客户端，和Master建立socket连接。
-
AcceptSocketService：Master服务端和Slave建立连接，并监听Slave的IO事件，建立HAConnection对象。
-
HAConnection：HAConnection是Master用来和Slave建立连接的类，处理和Slave的交互。Master（类似服务端）Slave（类似客户端）
-
ReadSocketService：ReadSocketService用来读取Slave向Master发送的数据，采用IO复用的方式处理。
-
WriteSocketService：WriteSocketService用来Master向Slave写返回的数据（commitlog的message数据）。
-
GroupTransferService： GroupTransferService是用来控制Master是否向Slave同步commitlog数据的。Master和Slave会进行通信，Master写message到内存ByteBuffer，然后调用handleHA()方法，然后构造一个同步请求放入GroupTransferService#requestsWrite的队列里，等待HAConnection#WriteSocketService处理这个请求,然后将commitlog的message数据，同步到Slave中。
图解类关系
HAClient类是Slave节点使用的，用来向Master通信的，相当于客户端的角色。
HAService是类Master节点使用的，里面包含了AcceptSocketService和GroupTransferService。
AcceptSocketService是同Slave建立连接，并监听Slave的IO事件，建立HAConnection连接对象。
HAConnection包含ReadSocketService和WriteSocketService；ReadSocketService用来读取Slave向Master发送的数据，采用IO复用的方式处理。WriteSocketService用来Master向Slave写返回的数据（commitlog的message数据）。
GroupTransferService是用来控制Master是否向Slave同步commitlog数据的。Master和Slave会进行通信，Master写message到内存ByteBuffer，然后调用handleHA()方法，然后构造一个同步请求放入GroupTransferService#requestsWrite的队列里，等待HAConnection#WriteSocketService处理这个请求,然后将commitlog的message数据，同步到Slave中。
image.png
Slave向Master通信
HAClient使Slave和Master建立连接，并报告自己同步的offset，然后等待Master的返回，并处理Master返回的message信息，写入到Slave本机的commitlog文件中，并构建consumequeue、indexfile索引文件。
@Override
public void run() {
log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
try {
// 连接Master
if (this.connectMaster()) {
// slave是否向Master发送offset消息,默认5秒发送一次
if (this.isTimeToReportOffset()) {
//Slave向Master发送当前Slave的commitlog的最大offset
boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset);
// 没有写完关闭Master
if (!result) {
this.closeMaster();
}
}
// selector使socketChannel等待1秒钟，等待
// I/O复用，检查是否有读事件
this.selector.select(1000);
// 处理Master返回的待处理消息，将返回的消息写入commitlog文件，并构建consumequeue、indexfile索引文件
boolean ok = this.processReadEvent();
if (!ok) {
// 关闭Master
this.closeMaster();
}
// 处理完读事件后，若slave的offset更新，需要再次发送新的slave的offset
if (!reportSlaveMaxOffsetPlus()) {
continue;
}
long interval =
HAService.this.getDefaultMessageStore().getSystemClock().now()
- this.lastWriteTimestamp;
if (interval > HAService.this.getDefaultMessageStore().getMessageStoreConfig()
.getHaHousekeepingInterval()) {
log.warn("HAClient, housekeeping, found this connection[" + this.masterAddress
+ "] expired, " + interval);
this.closeMaster();
log.warn("HAClient, master not response some time, so close connection");
}
} else {
// 连接失败，等待5秒;并不涉及线程之间的wait和notify操作等
this.waitForRunning(1000 * 5);
}
} catch (Exception e) {
log.warn(this.getServiceName() + " service has exception. ", e);
// 等待5秒，并不涉及线程之间的wait和notify操作等，然后再进行while循环，再次连接到master
this.waitForRunning(1000 * 5);
}
}
log.info(this.getServiceName() + " service end");
}
Master和Slave建立连接
AcceptSocketService是用来Master服务端和Slave建立连接，并监听Slave的IO事件，建立HAConnection对象。
/**
* Starts listening to slave connections.
*/
public void beginAccept() throws Exception {
this.serverSocketChannel = ServerSocketChannel.open();
this.selector = RemotingUtil.openSelector();
this.serverSocketChannel.socket().setReuseAddress(true);
// 绑定监听
this.serverSocketChannel.socket().bind(this.socketAddressListen);
// 非阻塞
this.serverSocketChannel.configureBlocking(false);
// serverSocketChannel注册到selector
this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
}
@Override
public void run() {
log.info(this.getServiceName() + " service started");
// 线程未停止
while (!this.isStopped()) {
try {
// 等待监听Socket的I/0完成事件通知，超时等待1秒
this.selector.select(1000);
// 被注册到selector上的key，也就是IO的socket
Set<SelectionKey> selected = this.selector.selectedKeys();
if (selected != null) {
// 遍历监听，一个监听事件一个HAConnection
for (SelectionKey k : selected) {
// 监听状态ok
if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
// 获取SocketChannel
SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();
if (sc != null) {
HAService.log.info("HAService receive new connection, "
+ sc.socket().getRemoteSocketAddress());
try {
// 建立和Slave的连接，里面包含了ReadSocketService用来读取Slave向Master发送的数据，WriteSocketService用来写Master向Slave返回的数据；
HAConnection conn = new HAConnection(HAService.this, sc);
// 开启ReadSocketService和WriteSocketService服务，处理Slave发来的请求和返回给Slave的数据
conn.start();
// 连接添加到集合
HAService.this.addConnection(conn);
} catch (Exception e) {
log.error("new HAConnection exception", e);
sc.close();
}
}
} else {
log.warn("Unexpected ops in select " + k.readyOps());
}
}
// 清空selected
selected.clear();
}
} catch (Exception e) {
log.error(this.getServiceName() + " service has exception.", e);
}
}
log.info(this.getServiceName() + " service end");
}
处理Slave发送的offset
ReadSocketService是Master用来读取Slave向Master发送的数据，采用IO复用的方式处理。
@Override
public void run() {
HAConnection.log.info(this.getServiceName() + " service started");
// 线程不会停止
while (!this.isStopped()) {
try {
// 同步轮询SocketChannel，等待IO事件通知完成，超时等待1秒
this.selector.select(1000);
// Master处理Slave发送的offset请求，并返回
boolean ok = this.processReadEvent();
if (!ok) {
HAConnection.log.error("processReadEvent error");
break;
}
long interval = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now() - this.lastReadTimestamp;
// Master和Slave连接超时间隔，20秒超时，记录log
if (interval > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaHousekeepingInterval()) {
log.warn("ha housekeeping, found this connection[" + HAConnection.this.clientAddr + "] expired, " + interval);
break;
}
} catch (Exception e) {
HAConnection.log.error(this.getServiceName() + " service has exception.", e);
break;
}
}
// Broker停机，线程关闭，资源释放
this.makeStop();
writeSocketService.makeStop();
haService.removeConnection(HAConnection.this);
HAConnection.this.haService.getConnectionCount().decrementAndGet();
SelectionKey sk = this.socketChannel.keyFor(this.selector);
if (sk != null) {
sk.cancel();
}
try {
this.selector.close();
this.socketChannel.close();
} catch (IOException e) {
HAConnection.log.error("", e);
}
HAConnection.log.info(this.getServiceName() + " service end");
}
// Master处理Slave发送的offset请求，并返回
private boolean processReadEvent() {
// 读取到数据为0byte的数据次数
int readSizeZeroTimes = 0;
// byteBufferRead不在包含空余空间，进行重新开启
if (!this.byteBufferRead.hasRemaining()) {
this.byteBufferRead.flip();
this.processPosition = 0;
}
// byteBufferRead还有剩余空间
while (this.byteBufferRead.hasRemaining()) {
try {
// 读取数据到byteBufferRead中
int readSize = this.socketChannel.read(this.byteBufferRead);
// 读取到数据
if (readSize > 0) {
// 更新readSizeZeroTimes和lastReadTimestamp
readSizeZeroTimes = 0;
this.lastReadTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
// byteBufferRead中读取到的数据位置>byteBufferRead上次处理过的数据>8;
// 读取超过8byte：8byte:代表slave向master发送的offset的大小8byte
if ((this.byteBufferRead.position() - this.processPosition) >= 8) {
// 获得slave发送的最大的offset的位置
int pos = this.byteBufferRead.position() - (this.byteBufferRead.position() % 8);
// 读取offset
long readOffset = this.byteBufferRead.getLong(pos - 8);
// 更新处理的位置
this.processPosition = pos;
// master接受到slave发送的offset
HAConnection.this.slaveAckOffset = readOffset;
if (HAConnection.this.slaveRequestOffset < 0) {
HAConnection.this.slaveRequestOffset = readOffset;
log.info("slave[" + HAConnection.this.clientAddr + "] request offset " + readOffset);
}
// 唤醒GroupTransferService#WaitNotifyObject#notifyTransferObject判断这个offset是否发送了，没有发送进行等待(GroupTransferService#notifyTransferObject.waitForRunning(1000))，
// 等待WriteSocketService写数据成功，然后再判断是否写入成功。
HAConnection.this.haService.notifyTransferSome(HAConnection.this.slaveAckOffset);
}
// 读取到数据的数据大小为0，3次跳出循环
} else if (readSize == 0) {
if (++readSizeZeroTimes >= 3) {
break;
}
} else {
log.error("read socket[" + HAConnection.this.clientAddr + "] < 0");
return false;
}
} catch (IOException e) {
log.error("processReadEvent exception", e);
return false;
}
}
return true;
}
Master向Slave写message消息
WriteSocketService用来Master向Slave写返回的数据（commitlog的message数据）。
@Override
public void run() {
HAConnection.log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
try {
// 同步轮询SocketChannel，等待IO事件通知完成，超时等待1秒
this.selector.select(1000);
// slave请求master的offset == -1，项目刚开始启动，master未接收到slave的拉取请求，sleep
if (-1 == HAConnection.this.slaveRequestOffset) {
Thread.sleep(10);
continue;
}
// nextTransferFromWhere = -1说明第一次进行数据传输，需要计算传输的物理偏移量
if (-1 == this.nextTransferFromWhere) {
// 如果slaveRequestOffset为0则从当前最后一个commitlog文件传输，否则根据slave broker的拉取请求偏移量开始
if (0 == HAConnection.this.slaveRequestOffset) {
// 确定Master的offset
long masterOffset = HAConnection.this.haService.getDefaultMessageStore().getCommitLog().getMaxOffset();
masterOffset =
masterOffset
- (masterOffset % HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig()
.getMappedFileSizeCommitLog());
if (masterOffset < 0) {
masterOffset = 0;
}
this.nextTransferFromWhere = masterOffset;
} else {
// 下次开始位置为slave请求位置
this.nextTransferFromWhere = HAConnection.this.slaveRequestOffset;
}
log.info("master transfer data from " + this.nextTransferFromWhere + " to slave[" + HAConnection.this.clientAddr
+ "], and slave request " + HAConnection.this.slaveRequestOffset);
}
// 上次写slave数据完成
if (this.lastWriteOver) {
// 距上次写数据间隔
long interval =
HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp;
if (interval > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig()
.getHaSendHeartbeatInterval()) {
// Build Header
this.byteBufferHeader.position(0);
this.byteBufferHeader.limit(headerSize);
this.byteBufferHeader.putLong(this.nextTransferFromWhere);
this.byteBufferHeader.putInt(0);
this.byteBufferHeader.flip();
this.lastWriteOver = this.transferData();
if (!this.lastWriteOver)
continue;
}
// 上次传输未结束则继续传输，可能是byteBufferHeader有剩余，也可能是SelectMappedBufferResult.ByteBuffer盛放消息的具体内容的数据还有剩余，没有被写完，重新开始写
} else {
this.lastWriteOver = this.transferData();
if (!this.lastWriteOver)
continue;
}
// 根据offset从master的commitlog文件获取数据
SelectMappedBufferResult selectResult =
HAConnection.this.haService.getDefaultMessageStore().getCommitLogData(this.nextTransferFromWhere);
if (selectResult != null) {
int size = selectResult.getSize();
if (size > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize()) {
size = HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize();
}
long thisOffset = this.nextTransferFromWhere;
this.nextTransferFromWhere += size;
selectResult.getByteBuffer().limit(size);
this.selectMappedBufferResult = selectResult;
// Build Header
this.byteBufferHeader.position(0);
this.byteBufferHeader.limit(headerSize);
this.byteBufferHeader.putLong(thisOffset);
this.byteBufferHeader.putInt(size);
this.byteBufferHeader.flip();
// 向slave的socket通道写数据
this.lastWriteOver = this.transferData();
} else {
// 如果没有获取到commitlog的数据，则进行等待；
// 一个Slave到Master的连接，一个HAConnection对象，一个WriteSocketService对象，一个线程，
// 因为Master没有最新的commitlog的数据，所以把所有的等待着数据的HAConnection的WriteSocketService()动作，进行等待；
// 将所有的HAConnection的WriteSocketService()线程被设置为未被通知的状态
HAConnection.this.haService.getWaitNotifyObject().allWaitForRunning(100);
}
} catch (Exception e) {
HAConnection.log.error(this.getServiceName() + " service has exception.", e);
break;
}
}
// 正常关机
// 将这个连接线程关闭，移除
HAConnection.this.haService.getWaitNotifyObject().removeFromWaitingThreadTable();
if (this.selectMappedBufferResult != null) {
this.selectMappedBufferResult.release();
}
this.makeStop();
readSocketService.makeStop();
haService.removeConnection(HAConnection.this);
SelectionKey sk = this.socketChannel.keyFor(this.selector);
if (sk != null) {
sk.cancel();
}
try {
this.selector.close();
this.socketChannel.close();
} catch (IOException e) {
HAConnection.log.error("", e);
}
HAConnection.log.info(this.getServiceName() + " service end");
}
// 向slave的socket通道写数据
private boolean transferData() throws Exception {
int writeSizeZeroTimes = 0;
// Write Header
// 如果读到Header数据的大小为0byte>3,跳出这个循环，进行下次header的写入
while (this.byteBufferHeader.hasRemaining()) {
int writeSize = this.socketChannel.write(this.byteBufferHeader);
if (writeSize > 0) {
writeSizeZeroTimes = 0;
this.lastWriteTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
} else if (writeSize == 0) {
if (++writeSizeZeroTimes >= 3) {
break;
}
} else {
throw new Exception("ha master write header error < 0");
}
}
// 根据offset从master的commitlog文件获取数据，maser是否有数据
if (null == this.selectMappedBufferResult) {
return !this.byteBufferHeader.hasRemaining();
}
writeSizeZeroTimes = 0;
// Write Body
// header被写满，开始写body；header：offset大小+4字节消息大小；header写满了，一定会有message的body，再去小body
if (!this.byteBufferHeader.hasRemaining()) {
while (this.selectMappedBufferResult.getByteBuffer().hasRemaining()) {
int writeSize = this.socketChannel.write(this.selectMappedBufferResult.getByteBuffer());
if (writeSize > 0) {
writeSizeZeroTimes = 0;
this.lastWriteTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
} else if (writeSize == 0) {
if (++writeSizeZeroTimes >= 3) {
break;
}
} else {
throw new Exception("ha master write body error < 0");
}
}
}
// header已经被写满 + selectMappedBufferResult里面存储message的内容的ByteBuffer已经被写完了，那这次写数据成功了。
boolean result = !this.byteBufferHeader.hasRemaining() && !this.selectMappedBufferResult.getByteBuffer().hasRemaining();
// 释放空间
if (!this.selectMappedBufferResult.getByteBuffer().hasRemaining()) {
this.selectMappedBufferResult.release();
this.selectMappedBufferResult = null;
}
return result;
}
Master和Slave同步通知控制
GroupTransferService是用来控制Master是否向Slave同步commitlog数据的。通过WaitNotifyObject来唤醒HAConnection中WriteSocketService向Slave写commitlog数据，同步等待5秒进行判断是否写入Slave是否成功。
Master和Slave会进行通信，Master写message到内存ByteBuffer，然后调用handleHA()方法，然后构造一个同步请求放入GroupTransferService#requestsWrite的队列里，等待HAConnection#WriteSocketService处理这个请求,然后将commitlog的message数据，同步到Slave中。
image.png
class GroupTransferService extends ServiceThread {
// 用来协调HAConnection中WriteSocketService和ReadSocketService之间的通信的
private final WaitNotifyObject notifyTransferObject = new WaitNotifyObject();
// 写请求队列，两个队列进行交换
private volatile List<CommitLog.GroupCommitRequest> requestsWrite = new ArrayList<>();
// 读请求队列，两个队列进行交换
private volatile List<CommitLog.GroupCommitRequest> requestsRead = new ArrayList<>();
// 放入请求到写队列
public synchronized void putRequest(final CommitLog.GroupCommitRequest request) {
synchronized (this.requestsWrite) {
this.requestsWrite.add(request);
}
// 唤醒处理这个Request的线程，唤醒doWaitTransfer()方法
this.wakeup();
}
// 通知Master的WriteSocketService给Slave传输一些数据
public void notifyTransferSome() {
this.notifyTransferObject.wakeup();
}
// 交换队列
private void swapRequests() {
List<CommitLog.GroupCommitRequest> tmp = this.requestsWrite;
this.requestsWrite = this.requestsRead;
this.requestsRead = tmp;
}
private void doWaitTransfer() {
synchronized (this.requestsRead) {
if (!this.requestsRead.isEmpty()) {
for (CommitLog.GroupCommitRequest req : this.requestsRead) {
// true，代表这个offset已经被推送过Slave了
boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
// request被处理的截止时间，消息从Master同步到Slave的同步等待时间5秒；
long waitUntilWhen = HAService.this.defaultMessageStore.getSystemClock().now()
+ HAService.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout();
// offset没有被推送过&&now<被处理的截止时间
// 图中是根据次数进行控制的，现在最新代码改为了时间判断
while (!transferOK && HAService.this.defaultMessageStore.getSystemClock().now() < waitUntilWhen) {
// WaitNotifyObject对象的waitForRunning(),交换读写队列，转变对象为未被通知的状态，并等待1秒，
// 等待WriteSocketService中将数据写入到Slave中，并更细push2SlaveMaxOffset，表示已经发送；具体发送动作在WriteSocketService中，这里只有一个判断是否发送成功，然后是等待，等待发送结果。
this.notifyTransferObject.waitForRunning(1000);
// push2SlaveMaxOffset被更新，大于request的offset，表示被Slave处理成功。
transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
}
if (!transferOK) {
log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
}
// 唤醒等待这个request处理结果的线程，应答存放这个request的线程，并返回结果；返回点为HandleHA()方法
req.wakeupCustomer(transferOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
}
this.requestsRead.clear();
}
}
}
public void run() {
log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
try {
// 父类ServiceThread的waitForRunning()方法，设置hasNotified为false，未被通知，然后交换写对队列和读队列，重置waitPoint为（1），休息200ms，finally设置hasNotified为未被通知，交换写对队列和读队列
this.waitForRunning(10);
//
this.doWaitTransfer();
} catch (Exception e) {
log.warn(this.getServiceName() + " service has exception. ", e);
}
}
log.info(this.getServiceName() + " service end");
}
@Override
protected void onWaitEnd() {
this.swapRequests();
}
@Override
public String getServiceName() {
return GroupTransferService.class.getSimpleName();
}
}
Message同步入库
Master和Slave会进行通信，Master写message到内存ByteBuffer，然后调用handleHA()方法，然后构造一个同步请求放入GroupTransferService#requestsWrite的队列里，等待HAConnection#WriteSocketService处理这个请求,然后将commitlog的message数据，同步到Slave中。
一个Slave到Master的连接，一个HAConnection对象，一个WriteSocketService对象，一个线程，将这个线程放入waitingThreadTable中，被设置这个线程未被通知的状态;
service.getWaitNotifyObject().wakeupAll();是唤醒所有等待的Master向Slave写CommitLog的message线程，向Slave同步数据。
/**
* commitlog的高可用，不同节点之间的构成commitlog的message复制，每条消息进行一次方法调用
* @param result 追加消息到ByteBuffer中的返回结果
* @param putMessageResult 放入ByteBuffer这个过程的结果（存放消息的结果）
* @param messageExt 需要存放的消息
*/
public void handleHA(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
// 是同步Master的角色
if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
// HA服务
HAService service = this.defaultMessageStore.getHaService();
// 是否等待消息落盘完毕
if (messageExt.isWaitStoreMsgOK()) {
// Determine whether to wait
// 推送这条消息，Slave是否可以接受这条消息推送
if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
// 构建Master到Slave的同步请求
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
// 放入请求，并唤醒处理这个request的服务：HAService#GroupTransferService的doWaitTransfer()方法，处理这个request。
service.putRequest(request);
// HAService#WaitNotifyObject对象
// 一个Slave到Master的连接，一个HAConnection对象，一个WriteSocketService对象，一个线程，将这个线程放入waitingThreadTable中，被设置这个线程未被通知的状态;
// 这里的作用是唤醒所有等待的Master向Slave写CommitLog的message线程，向Slave同步数据。
service.getWaitNotifyObject().wakeupAll();
PutMessageStatus replicaStatus = null;
try {
// 同步等待写入Slave的commitlog消息返回结果，超时等待5秒
replicaStatus = request.future().get(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout(),
TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
}
// 同步失败，记录log
if (replicaStatus != PutMessageStatus.PUT_OK) {
log.error("do sync transfer other node, wait return, but failed, topic: " + messageExt.getTopic() + " tags: "
+ messageExt.getTags() + " client address: " + messageExt.getBornHostNameString());
putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
}
}
// Slave problem
// 存放Slave不可用结果，并返回
else {
// Tell the producer, slave not available
putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE);
}
}
}
}
线程通知关键
- GroupTransferService#WaitNotifyObject#notifyTransferObject
HAConnection的WriteSocketService和ReadSocketService都没有使用ThreadService的通知和等待系统，WriteSocketService和ReadSocketService通信采用了GroupTransferService#WaitNotifyObject#notifyTransferObject这个协调两个线程之间的通信。
- HAService#WaitNotifyObject#waitNotifyObject唤醒
HAService#WaitNotifyObject#waitNotifyObject的唤醒
是用来通知所有的HAConnection的WriteSocketService中Master向Slave写CommitLog数据的，
调用的地方CommitLog#handleHA()#service.getWaitNotifyObject().wakeupAll()。
- HAService#WaitNotifyObject#waitNotifyObject等待
HAService#WaitNotifyObject#waitNotifyObject等待
是用来停止所有的HAConnection的WriteSocketService中Master向Slave写CommitLog数据的，
调用的地方是在：HAConnection的#WriteSocketService#HAConnection.this.haService.getWaitNotifyObject().allWaitForRunning(100);
- GroupTransferService#ThredService父类
GroupTransferService的ServiceThread的唤醒和等待是用来处理本线程读队列和写队列之间处理同步数据请求的，和同步刷盘服务GroupCommitService一样的效果。
