当前位置:首页 > 开发 > 开源软件 > 正文

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

发表于: 2014-08-25   作者:asialee   来源:转载   浏览次数:
摘要:          1. QueueItem介绍,从类的定义来看,这个里面包含一个数据,超那个topic的那个分区发送           public class QueueItem<T> { public final T data ; public fi

         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);
        }
    }
 
        

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

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
首先我们来看看producer的例子: 首先我们先看单元测试ProducerTest这个类里面的发送消息的代码: @
首先我们来看看producer的例子: 首先我们先看单元测试ProducerTest这个类里面的发送消息的代码: @
Jafka类为整个程序的入口类, 这个类本身没有什么特别的,里面有两个小的知识点需要注意下。 在启动
RequestHandler从代码来看是比较简单的一个组件,在producer和consumer和broker打交道的过程中,均
Send这块是比较好理解的,它的作用就是produer或者consumer给broker发消息后的响应,先来看下类图。
今天终于要讲到LogManager了,在讲LogManager之前,我们还是先来看下几个基本概念和jafka的存储结构
Jafka类为整个程序的入口类, 这个类本身没有什么特别的,里面有两个小的知识点需要注意下。 在启动
RequestHandler从代码来看是比较简单的一个组件,在producer和consumer和broker打交道的过程中,均
Send这块是比较好理解的,它的作用就是produer或者consumer给broker发消息后的响应,先来看下类图。
今天终于要讲到LogManager了,在讲LogManager之前,我们还是先来看下几个基本概念和jafka的存储结构
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号