一、Topic的相关操作和可视化工具

手动创建Topic

看过SpringBoot-Kafka（一）这篇文章会发现，我们从到到尾都没有创建过"topic.quick.demo"这个Topic，这是因为KafkaTemplate在发送的时候就已经帮我们完成了创建的操作，所以我们不需要主动创建"topic.quick.demo"这个Topic，而是交由KafkaTemplate去完成。但这样也出现了问题，这种情况创建出来的Topic的Partition(分区)数是默认值（3），副本数也是默认值（1），这就导致了我们在后期不能顺利扩展。所以这种情况我们需要使用代码手动去创建Topic。

可视化工具Kafka Tool 2

Kafka Tool 2是一款Kafka的可视化客户端工具，可以非常方便的查看Topic的队列信息以及消费者信息以及kafka节点信息。直接丢下载地址：http://www.kafkatool.com/download.html

打开之后我们看到的界面如下，非常简洁，虽然这个工具没有诸如KafkaOffsetMonitor这种监控工具的功能强大，但胜在操作方便，后期会补充一下监控工具的使用。

主界面，我们点击左上角的新建连接按钮去创建连接

创建连接
接下来会弹出一个配置连接的窗口，我们可以看到这个窗口左上角为Add Cluster(添加集群)，但没关系，对应单节点的Kafka实例来说也是可以的，因为这个软件监控的是Zookeeper而不是Kafka，Kafka的集群搭建也是依赖Zookeeper来实现的，所以默认情况下我们都是直接通过Zookeeper去完成大部分操作。

配置连接
点击Add后我们可以看到已经创建好的Topic。这个软件默认显示数据的类型为Byte，说白了我们是不能直接看到消息的明文数据的，没关系，我们可以在设置里面找到对应的修改选项，接下来就自己探索吧

使用@Bean注解创建Topic

可以说使用SpringBoot创建Topic是一件非常简单的事情。
首先我们在config包下创建KafkaInitialConfiguration类，注册一个类型为NewTopic的Bean即可。

@Configuration
public class KafkaInitialConfiguration {

    //创建TopicName为topic.quick.initial的Topic并设置分区数为8以及副本数为1
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic("topic.quick.initial",8, (short) 1 );
    }
}

接下来启动一下SpringBoot项目，启动完成后打开Kafka Tool 2工具，查看一下刚才创建的队列是否存在

手动编码创建Topic

同样在KafkaInitialConfiguration类中编码，注册KafkaAdmin和AdminClient两个Bean

@Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> props = new HashMap<>();
        //配置Kafka实例的连接地址
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        KafkaAdmin admin = new KafkaAdmin(props);
        return admin;
    }

    @Bean
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin().getConfig());
    }

接下来在DemoTest测试类中编写测试方法，这里需要注意一点Topic的新增删除方法都是异步执行的，为了避免在创建过程中程序关闭导致创建失败，所以在代码最后加了一秒的休眠，执行测试方法我们打开Kafka Tool 2会发现多出了一个"topic.quick.initial2"的Topic

    @Autowired
    private AdminClient adminClient;

    @Test
    public void testCreateTopic() throws InterruptedException {
        NewTopic topic = new NewTopic("topic.quick.initial2", 1, (short) 1);
        adminClient.createTopics(Arrays.asList(topic));
        Thread.sleep(1000);
    }

修改Topic分区数量

为什么要更新Topic呢，例如我们上一章创建的“topic.qucik.demo”只有一个分区，后期我们想增加分区数来提高系统吞吐量，这样我们就需要修改一下Topic的分区数了。实现也非常简单，只需要修改在我们刚才编写的KafkaInitialConfiguration类的initialTopic()方法，紧接着重启一下项目即可。修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小。

    @Bean
    public NewTopic initialTopic() {
        return new NewTopic("topic.quick.initial",8, (short) 1 );
    }
    //修改后|
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic("topic.quick.initial",11, (short) 1 );
    }

修改成功

查询Topic信息

测试类中新建一个testSelectTopicInfo方法 ，使用lambda表达式遍历输出

    @Test
    public void testSelectTopicInfo() throws ExecutionException, InterruptedException {
        DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList("topic.quick.initial"));
        result.all().get().forEach((k,v)->System.out.println("k: "+k+" ,v: "+v.toString()+"\n"));
    }

这个是输出结果，里面就包含了各个分区的信息等等

k: topic.quick.initial ,v: (name=topic.quick.initial,
    internal=false,
    partitions=(partition=0, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=1, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=2, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=3, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=4, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=5, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=6, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=7, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=8, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=9, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=10, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)))

二、KafkaTemplate发送消息及结果回调

我们使用KafkaTemplate.send(String data)这个方法发送消息到Kafka中，显然这个方法并不能满足我们系统的需求，那我们需要查看一下KafkaTemplate所实现的接口，看看还提供了什么方法。当我们发送消息到Kafka后，我们又怎么去确认消息是否发送成功呢？这就涉及到KafkaTemplate的发送回调方法了。接下来我们开始正式讲解。

查看发送接口

在KafkaTemplate的源代码中，可以看到有关发送的接口如下。这里的参数还是比较简单的，值得一提的事，方法中有个Long类型的时间戳(timestamp)参数，这是Kafka0.10版本提供的新功能，主要用来使用时间索引进行查询数据以及日志切分清除策略。还有一个ProducerRecord参数，这个类其实就是整合了topic、partition、data等数据的消费实体类。
参数意思：
topic：这里填写的是Topic的名字
partition：这里填写的是分区的id，其实也是就第几个分区，id从0开始。表示指定发送到该分区中
timestamp：时间戳，一般默认当前时间戳
key：消息的键
data：消息的数据
ProducerRecord：消息对应的封装类，包含上述字段
Message<?>：Spring自带的Message封装类，包含消息及消息头

ListenableFuture<SendResult<K, V>> sendDefault(V data);

ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);

ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);

ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);

ListenableFuture<SendResult<K, V>> send(String topic, V data);

ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);

ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);

ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);

ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);

ListenableFuture<SendResult<K, V>> send(Message<?> message);

使用sendDefault发送消息

首先在KafkaConfiguration编写一个带有默认Topic参数的KafkaTemplate，同时为另外一个KafkaTemplate加上@Primary注解，@Primary注解的意思是在拥有多个同类型的Bean时优先使用该Bean，到时候方便我们使用@Autowired注解自动注入。

    //这个是我们之前编写的KafkaTemplate代码，加入@Primary注解
    @Bean
    @Primary
    public KafkaTemplate<Integer, String> kafkaTemplate() {
        KafkaTemplate template = new KafkaTemplate<Integer, String>(producerFactory());
        return template;
    }

    @Bean("defaultKafkaTemplate")
    public KafkaTemplate<Integer, String> defaultKafkaTemplate() {
        KafkaTemplate template = new KafkaTemplate<Integer, String>(producerFactory());
        template.setDefaultTopic("topic.quick.default");
        return template;
    }

接着编写测试方法，可以看到我们这里调用的是sendDefault方法，而且并没有在方法参数上添加topicName，这是因为我们在声明defaultKafkaTemplate这个Bean的时候添加了这行代码 template.setDefaultTopic("topic.quick.default")，只要调用sendDefault方法，kafkaTemplate会自动把消息发送到名为"topic.quick.default"的Topic中。

    @Resource
    private KafkaTemplate defaultKafkaTemplate;

    @Test
    public void testDefaultKafkaTemplate() {
        defaultKafkaTemplate.sendDefault("I`m send msg to default topic");
    }

其他方法使用：

    @Test
    public void testTemplateSend() {
        //发送带有时间戳的消息
        kafkaTemplate.send("topic.quick.demo", 0, System.currentTimeMillis(), 0, "send message with timestamp");

        //使用ProducerRecord发送消息
        ProducerRecord record = new ProducerRecord("topic.quick.demo", "use ProducerRecord to send message");
        kafkaTemplate.send(record);

        //使用Message发送消息
        Map map = new HashMap();
        map.put(KafkaHeaders.TOPIC, "topic.quick.demo");
        map.put(KafkaHeaders.PARTITION_ID, 0);
        map.put(KafkaHeaders.MESSAGE_KEY, 0);
        GenericMessage message = new GenericMessage("use Message to send message",new MessageHeaders(map));
        kafkaTemplate.send(message);
    }

消息结果回调

一般来说我们都会去获取KafkaTemplate发送消息的结果去判断消息是否发送成功，如果消息发送失败，则会重新发送或者执行对应的业务逻辑。所以这里我们去实现这个功能。

KafkaSendResultHandler

第一步还是编写一个消息结果回调类KafkaSendResultHandler。当我们使用KafkaTemplate发送消息成功的时候回调用OnSuccess方法，发送失败则会调用onError方法。

@Component
public class KafkaSendResultHandler implements ProducerListener {

    private static final Logger log = LoggerFactory.getLogger(KafkaSendResultHandler.class);

    @Override
    public void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata) {
        log.info("Message send success : " + producerRecord.toString());
    }

    @Override
    public void onError(ProducerRecord producerRecord, Exception exception) {
        log.info("Message send error : " + producerRecord.toString());
    }
}

接下来就使用KafkaSendResultHandler实现消息发送结果回调

    @Autowired
    private KafkaSendResultHandler producerListener;

    @Test
    public void testProducerListen() throws InterruptedException {
        kafkaTemplate.setProducerListener(producerListener);
        kafkaTemplate.send("topic.quick.demo", "test producer listen");
        Thread.sleep(1000);
    }

运行测试方法，我们可以看到控制台输出的日志如下

2018-09-08 15:51:39.975  INFO 10268 --- [ad | producer-1] c.v.k.handler.KafkaSendResultHandler     : Message send success : ProducerRecord(topic=topic.quick.demo, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=null, value=test producer listen, timestamp=null)

KafkaTemplate异步发送消息

上文提及了发送消息的时候需要休眠一下，否则发送时间较长的时候会导致进程提前关闭导致无法调用回调时间。主要是因为KafkaTemplate发送消息是采取异步方式发送的，我们可以看下KafkaTemplate的源代码，从刚才调用的发送消息方法，可以看到KafkaTemplate会使用ProducerRecord把我们传递进来的参数再一次封装，最后调用doSend方法发送消息到Kafka中

send(String topic, V data)

    public ListenableFuture<SendResult<K, V>> send(String topic, V data) {
        ProducerRecord<K, V> producerRecord = new ProducerRecord(topic, data);
        return this.doSend(producerRecord);
    }

ListenableFuture<SendResult<K, V>> doSend(...)

doSend方法先是检测是否开启事务，紧接着使用SettableListenableFuture发送消息，然后判断是否启动自动冲洗数据到Kafka中，我们再接着看看SettableListenableFuture实现了什么接口

    protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord) {
        if (this.transactional) {
            Assert.state(this.inTransaction(), "No transaction is in process; possible solutions: run the template operation within the scope of a template.executeInTransaction() operation, start a transaction with @Transactional before invoking the template method, run in a transaction started by a listener container when consuming a record");
        }

        final Producer<K, V> producer = this.getTheProducer();
        if (this.logger.isTraceEnabled()) {
            this.logger.trace("Sending: " + producerRecord);
        }

        final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture();
        producer.send(producerRecord, new Callback() {
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                try {
                    if (exception == null) {
                        future.set(new SendResult(producerRecord, metadata));
                        if (KafkaTemplate.this.producerListener != null) {
                            KafkaTemplate.this.producerListener.onSuccess(producerRecord, metadata);
                        }

                        if (KafkaTemplate.this.logger.isTraceEnabled()) {
                            KafkaTemplate.this.logger.trace("Sent ok: " + producerRecord + ", metadata: " + metadata);
                        }
                    } else {
                        future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception));
                        if (KafkaTemplate.this.producerListener != null) {
                            KafkaTemplate.this.producerListener.onError(producerRecord, exception);
                        }

                        if (KafkaTemplate.this.logger.isDebugEnabled()) {
                            KafkaTemplate.this.logger.debug("Failed to send: " + producerRecord, exception);
                        }
                    }
                } finally {
                    if (!KafkaTemplate.this.transactional) {
                        KafkaTemplate.this.closeProducer(producer, false);
                    }

                }

            }
        });
        if (this.autoFlush) {
            this.flush();
        }

        if (this.logger.isTraceEnabled()) {
            this.logger.trace("Sent: " + producerRecord);
        }

        return future;
    }

可以看到SettableListenableFuture实现了ListenableFuture接口，ListenableFuture则实现了Future接口，Future是Java自带的实现异步编程的接口，支持返回值的异步，而我们使用Thread或者Runnable都是不带返回值的。

public class SettableListenableFuture<T> implements ListenableFuture<T>
public interface ListenableFuture<T> extends Future<T> 

KafkaTemplate同步发送消息

KafkaTemplate异步发送消息大大的提升了生产者的并发能力，但某些场景下我们并不需要异步发送消息，这个时候我们可以采取同步发送方式，实现也是非常简单的，我们只需要在send方法后面调用get方法即可。Future模式中，我们采取异步执行事件，等到需要返回值得时候我们再调用get方法获取future的返回值

    @Test
    public void testSyncSend() throws ExecutionException, InterruptedException {
        kafkaTemplate.send("topic.quick.demo", "test sync send message").get();
    }

get方法还有一个比较有意思的重载方法，get(long timeout, TimeUnit unit)，当send方法耗时大于get方法所设定的参数时会抛出一个超时异常，但需要注意，这里仅抛出异常，消息还是会发送成功的。这里的测试方法设置send耗时必须小于 一微秒，运行后我们可以看到抛出的异常，但也发现消息能发送成功并被监听器接收了。那这功能有什么作用呢，如果还没有接触过SQL慢查询可以去了解一下，使用该方法作为SQL慢查询记录的条件。

    @Test
    public void testTimeOut() throws ExecutionException, InterruptedException, TimeoutException {
        kafkaTemplate.send("topic.quick.demo", "test send message timeout").get(1,TimeUnit.MICROSECONDS);
    }

2018-09-08 16:36:09.110  INFO 7724 --- [     demo-0-C-1] com.viu.kafka.listen.DemoListener        : demo receive : test send message timeout

java.util.concurrent.TimeoutException