#解压文件
tar -zxf kafka_2.11-1.1.0.tgz
cd kafka_2.11-1.1.0

#启动Zookeerper
bin/zookeeper-server-start.sh config/zookeeper.properties

#启动Kafka
bin/kafka-server-start.sh config/server.properties &
cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties

config/server-1.properties:
    broker.id=1
    listeners=PLAINTEXT://:9093
    log.dir=/tmp/kafka-logs-1
 
config/server-2.properties:
    broker.id=2
    listeners=PLAINTEXT://:9094
    log.dir=/tmp/kafka-logs-2

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

#创建集群
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic myTopic

#查看主题
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic myTopic

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

}