package cn.codepeople.config;

import java.util.Date;
import java.util.UUID;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import com.alibaba.fastjson.JSONObject;

import cn.codepeople.dto.KafkaVO;


/**
 * kafka消息的生产者,消息的发送方
 * @author 刘仁
 *
 */
@Component
@EnableScheduling
public class KafkaProducer {

	@Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 定时任务
     */
    @Scheduled(cron = "00/1 * * * * ?")
    public void send(){
        String message = UUID.randomUUID().toString();
        KafkaVO vo = new KafkaVO();
        vo.setId(message);
        vo.setMsg("需要发送消息的内容");
        vo.setDate(new Date());
        ListenableFuture<?> future = kafkaTemplate.send("app_log",JSONObject.toJSONString(vo));
        String message2 = "第二种类的订阅消息发送";
        ListenableFuture<?> future2 = kafkaTemplate.send("app_log2",message2);
        future.addCallback(o -> System.out.println("send-消息发送成功：" + message), throwable -> System.out.println("消息发送失败：" + message));
        future2.addCallback(o -> System.out.println("send-消息发送成功：" + message2), throwable -> System.out.println("消息发送失败：" + message2));
    }
}

package cn.codepeople.config;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
 * kafka消息的消费者
 * @author 刘仁
 *
 */
@Component
public class KafkaConsumer {

	@KafkaListener(topics = {"app_log"})
    public void receive(String message){
        System.out.println("app_log--消费消息:" + message);
    }
	
	@KafkaListener(topics = {"app_log2"})
    public void receive2(String message){
        System.out.println("app_log2--消费消息:" + message);
    }
}

/**   
 * @author lr
 * @date 2019年3月4日 上午11:30:02 
 * @version V1.0.0   
 */
package cn.codepeople;

import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SysApplication {

    public static void main(String[] args) {
        SpringApplication.run(SysApplication.class, args);
    }
}