#解压文件

tar -zxf kafka_2.-1.1..tgz

cd kafka_2.-1.1.

#启动Zookeerper

bin/zookeeper-server-start.sh config/zookeeper.properties

#启动Kafka

bin/kafka-server-start.sh config/server.properties &

cp config/server.properties config/server-.properties

cp config/server.properties config/server-.properties

config/server-.properties:

    broker.id=

    listeners=PLAINTEXT://:

    log.dir=/tmp/kafka-logs-

config/server-.properties:

    broker.id=

    listeners=PLAINTEXT://:

    log.dir=/tmp/kafka-logs-

bin/kafka-server-start.sh config/server-.properties &

bin/kafka-server-start.sh config/server-.properties &

#创建集群

bin/kafka-topics.sh --create --zookeeper localhost: --replication-factor  --partitions  --topic myTopic

#查看主题

bin/kafka-topics.sh --describe --zookeeper localhost: --topic myTopic

<dependency>

    <groupId>org.springframework.kafka</groupId>

    <artifactId>spring-kafka</artifactId>

</dependency>

spring:

  kafka:

    bootstrap-servers: 10.123.52.76:9092,10.123.52.76:9093,10.123.52.76:9094

    consumer:

      group-id: myGroup

package com.cjs.boot.message;

import lombok.extern.slf4j.Slf4j;

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.stereotype.Component;

@Slf4j

@Component

public class MyListener {

    @KafkaListener(topics = "myTopic")

    public void processMessage2(String content) {

        log.info("【Received Message From 'myTopic'】: {}", content);

    }

}

package com.cjs.boot.controller;

import com.cjs.boot.response.RespResult;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.kafka.core.KafkaTemplate;

import org.springframework.kafka.support.SendResult;

import org.springframework.stereotype.Controller;

import org.springframework.util.concurrent.ListenableFuture;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.PostMapping;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.ResponseBody;

import org.springframework.web.servlet.ModelAndView;

@Controller

@RequestMapping("/message")

public class MessageController extends BaseController {

    @Autowired

    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/add.html")

    public ModelAndView add() {

        return new ModelAndView("message/add");

    }

    @PostMapping("/send.json")

    @ResponseBody

    public RespResult send(String text) {

        ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send("myTopic", String.valueOf(System.currentTimeMillis()), text);

        return RespResult.success();

    }

}

2018-05-04 12:36:59.736  INFO 7552 --- [nio-8080-exec-2] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 1.0.1

2018-05-04 12:36:59.736  INFO 7552 --- [nio-8080-exec-2] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : c0518aa65f25317e2018-05-04 12:36:59.830  INFO 7552 --- [ntainer#0-0-C-1] com.cjs.boot.message.MyListener          : 【Received Message From 'myTopic'】: 大家好啊

2018-05-04 12:37:24.107  INFO 7552 --- [ntainer#0-0-C-1] com.cjs.boot.message.MyListener          : 【Received Message From 'myTopic'】: 吃饭啦