Yahoo pulsar -1 Producer send和Consumer receive 过程源码剖析

本文主要是介绍Yahoo pulsar -1 Producer send和Consumer receive 过程源码剖析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1、同步发送消息

MessageId send(byte[] message) throws PulsarClientException;

private static final AtomicLongFieldUpdater<ProducerImpl> msgIdGeneratorUpdater = AtomicLongFieldUpdater
.newUpdater(ProducerImpl.class, "msgIdGenerator");

msgIdGeneratorUpdater 用于生成请求ID,每次发送请求ID都不一样,用来区分消息。

private final BlockingQueue<OpSendMsg> pendingMessages = Queues.newArrayBlockingQueue(conf.getMaxPendingMessages()); 阻塞的消息队列,用于缓存发送消息。

从Channel所在的线程 eventLoop()来执行发送请求。避免线程切换。

// Read the connection before validating if it's still connected, so that we avoid reading a null
// value
ClientCnx cnx = cnx();
if (isConnected()) {
// If we do have a connection, the message is sent immediately, otherwise we'll try again once a
// new
// connection is established
cmd.retain();
cnx.ctx().channel().eventLoop().execute(WriteInEventLoopCallback.create(this, cnx, op));

}

    private static final class WriteInEventLoopCallback implements Runnable {private ProducerImpl producer;private ClientCnx cnx;private OpSendMsg op;static WriteInEventLoopCallback create(ProducerImpl producer, ClientCnx cnx, OpSendMsg op) {WriteInEventLoopCallback c = RECYCLER.get();c.producer = producer;c.cnx = cnx;c.op = op;return c;}@Overridepublic void run() {if (log.isDebugEnabled()) {log.debug("[{}] [{}] Sending message cnx {}, sequenceId {}", producer.topic, producer.producerName, cnx,op.sequenceId);}try {cnx.ctx().writeAndFlush(op.cmd, cnx.ctx().voidPromise());} finally {recycle();}}private void recycle() {producer = null;cnx = null;op = null;RECYCLER.recycle(this, recyclerHandle);}private final Handle recyclerHandle;private WriteInEventLoopCallback(Handle recyclerHandle) {this.recyclerHandle = recyclerHandle;}private static final Recycler<WriteInEventLoopCallback> RECYCLER = new Recycler<WriteInEventLoopCallback>() {@Overrideprotected WriteInEventLoopCallback newObject(Handle handle) {return new WriteInEventLoopCallback(handle);}};}

发送消息后接受响应消息：