Jafka学习之Producer发送之组件说明

         1. QueueItem介绍，从类的定义来看，这个里面包含一个数据，超那个topic的那个分区发送

         

public class QueueItem<T> {

    public final T data ;

    public final int partition ;

    public final String topic ;
}

         2. EventHandler介绍，从类的定义来看，可以初始化，可以被关闭，在handle方法里面包含一个encoder，一个SyncProducer，和一批消息

      

public interface EventHandler<T> extends Closeable{

    /**
     * Initializes the event handler using a Properties object
     *
     * @param properties the properties used to initialize the event
     *        handler
     */
    void init(Properties properties);

    /**
     * Callback to dispatch the batched data and send it to a Jafka server
     *
     * @param events the data sent to the producer
     * @param producer the low- level producer used to send the data
     * @param encoder data encoder
     */
    void handle(List<QueueItem<T>> events, SyncProducer producer, Encoder<T> encoder);

    /**
     * Cleans up and shuts down the event handler
     */
    void close();
}

 

           3. CallBackHandler从类的命名来看就是一个回调函数，这个在异步发送的时候，可以回调用户端的一些程序

         

public interface EventHandler<T> extends Closeable{

    /**
     * Initializes the event handler using a Properties object
     *
     * @param properties the properties used to initialize the event
     *        handler
     */
    void init(Properties properties);

    /**
     * Callback to dispatch the batched data and send it to a Jafka server
     *
     * @param events the data sent to the producer
     * @param producer the low- level producer used to send the data
     * @param encoder data encoder
     */
    void handle(List<QueueItem<T>> events, SyncProducer producer, Encoder<T> encoder);

    /**
     * Cleans up and shuts down the event handler
     */
    void close();
}

          4 我们接着来看下SyncProducer的实现

        

public class SyncProducer implements Closeable {

    private final Logger logger = Logger.getLogger(SyncProducer .class );

    //private static final RequestKeys RequestKey = RequestKeys.Produce;//0

    /////////////////////////////////////////////////////////////////////
    // 同步发送器的配置
    private final SyncProducerConfig config ;
    // BlockingChannel这个待会再讲，感觉是封装了一个channel
    private final BlockingChannel blockingChannel ;
    // 这个是一个对象锁，待会讲解锁的妙用
    private final Object lock = new Object();
    // 是否已经关闭的标志位
    private volatile boolean shutdown = false;

    // 主机和端口号
    private final String host ;

    private final int port ;

  接着是它的构造函数：
 public SyncProducer(SyncProducerConfig config) {
        super();
        this.config = config;
        this.host = config.getHost();
        this.port = config.getPort();
        //
        this.blockingChannel = new BlockingChannel(host, port , -1, config.socketTimeoutMs, config.bufferSize );
    }

        这个里面讲readBufferSize设置成-1了，因为它不需要读数据

        我们接下来看下这三个send的重载函数

        

// 采用随机partition进行发送
public void send(String topic, ByteBufferMessageSet message) {
        send(topic, ProducerRequest. RandomPartition, message);
    }

// 检验消息大小后，构建ProducerRequest对象进行send
    public void send(String topic, int partition, ByteBufferMessageSet messages) {
        messages.verifyMessageSize(config .maxMessageSize );
        send( new ProducerRequest(topic, partition, messages));
    }

    private void send(Request request) {
        // 从request对象构建出send对象
        BoundedByteBufferSend send = new BoundedByteBufferSend(request);
        synchronized (lock ) {
            long startTime = System.nanoTime();
            int written = -1;
            try {
                // 建立链接并发送send，最终返回一个number，表明发送的字节数量
                written = connect().send(send);
            } catch (IOException e) {
                // no way to tell if write succeeded. Disconnect and re-throw exception to let client handle retry
                disconnect();
                throw new RuntimeException(e);
            } finally {
                if (logger .isDebugEnabled()) {
                    logger.debug(format( "write %d bytes data to %s:%d", written, host, port));
                }
            }
            final long endTime = System.nanoTime ();
            SyncProducerStats.recordProduceRequest(endTime - startTime);
        }
    }