springboot-kafka

Author: Wenfeng.Huang

springboot-kafka/.gitattributes

@@ -0,0 +1,5 @@
+*.js linguist-language=java
+*.html linguist-language=java
+*.java linguist-language=java
+*.xml linguist-language=java
+*.sh linguist-language=java

springboot-kafka/README.md

@@ -0,0 +1,21 @@
+1.下载kafka安装包、并安装：
+
+ 下载地址：http://kafka.apache.org/
+ 
+  
+ 安装:
+ tar -zxvf kafka_2.11-0.11.0.0.tgz
+2.修改kafka配置文件 server.properties 加入如下配置：
+
+ listeners=PLAINTEXT://192.168.0.195:9092 #使其他主机访问
+ auto.create.topics.enable = true #自动创建topic
+
+3.开启9092端口，使外部可以访问。
+
+4.启动kafka：
+
+ cd kafka_2.11-0.11.0.0
+ bin/zookeeper-server-start.sh config/zookeeper.properties & # 先启动zookeeper
+ bin/kafka-server-start.sh config/server.properties & # 启动kafka服务
+
+5.打开项目选择KafkaApplication并运行。

springboot-kafka/pom.xml

@@ -0,0 +1,67 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<modelVersion>4.0.0</modelVersion>
+
+<groupId>com.chen</groupId>
+<artifactId>kafka</artifactId>
+<version>0.0.1-SNAPSHOT</version>
+<packaging>jar</packaging>
+
+<name>kafka</name>
+<description>Demo project for Spring Boot</description>
+
+<parent>
+<groupId>org.springframework.boot</groupId>
+<artifactId>spring-boot-starter-parent</artifactId>
+<version>1.5.6.RELEASE</version>
+<relativePath/> <!-- lookup parent from repository -->
+</parent>
+
+<properties>
+<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+<java.version>1.8</java.version>
+</properties>
+
+<dependencies>
+<dependency>
+<groupId>org.springframework.kafka</groupId>
+<artifactId>spring-kafka</artifactId>
+</dependency>
+<dependency>
+<groupId>org.springframework.boot</groupId>
+<artifactId>spring-boot-starter-thymeleaf</artifactId>
+</dependency>
+<dependency>
+<groupId>org.springframework.boot</groupId>
+<artifactId>spring-boot-starter-web</artifactId>
+</dependency>
+<!-- https://mvnrepository.com/artifact/com.google.code.gson/gson -->
+<dependency>
+<groupId>com.google.code.gson</groupId>
+<artifactId>gson</artifactId>
+</dependency>
+<dependency>
+<groupId>org.projectlombok</groupId>
+<artifactId>lombok</artifactId>
+<optional>true</optional>
+</dependency>
+<dependency>
+<groupId>org.springframework.boot</groupId>
+<artifactId>spring-boot-starter-test</artifactId>
+<scope>test</scope>
+</dependency>
+</dependencies>
+
+<build>
+<plugins>
+<plugin>
+<groupId>org.springframework.boot</groupId>
+<artifactId>spring-boot-maven-plugin</artifactId>
+</plugin>
+</plugins>
+</build>
+
+
+</project>

springboot-kafka/src/main/java/com/chen/KafkaApplication.java

@@ -0,0 +1,29 @@
+package com.chen;
+
+import com.chen.common.util.ToolsUtil;
+import com.chen.common.message.PayMessage;
+import com.chen.producer.MessageProducer;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ApplicationContext;
+
+@SpringBootApplication
+public class KafkaApplication {
+
+ public static void main(String[] args) {
+ ApplicationContext applicationContext = SpringApplication.run(KafkaApplication.class, args);
+ MessageProducer producer = applicationContext.getBean(MessageProducer.class);
+ while (true){
+ PayMessage message = new PayMessage();
+ message.setFee(ToolsUtil.getFee());
+ message.setOrderCode(ToolsUtil.getNextCode());
+ message.setSendTime(System.currentTimeMillis());
+ producer.send(message);
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+}

springboot-kafka/src/main/java/com/chen/common/constant/TopicConst.java

@@ -0,0 +1,13 @@
+package com.chen.common.constant;
+
+/**
+ * 类的功能描述：
+ * 定义topic名称
+ *
+ * @ClassName: TopicConst
+ * @Author hcxin
+ * @Date 2017-09-13 03:49:22
+ */
+public class TopicConst {
+ public static final String PAY_TOPIC = "payTopic";
+}

springboot-kafka/src/main/java/com/chen/common/message/PayMessage.java

@@ -0,0 +1,16 @@
+package com.chen.common.message;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+import java.util.Date;
+
+@Getter
+@Setter
+@ToString
+public class PayMessage {
+ private String orderCode;
+ private Float fee;
+ private Long sendTime;
+}

springboot-kafka/src/main/java/com/chen/common/util/ToolsUtil.java

@@ -0,0 +1,29 @@
+package com.chen.common.util;
+
+import java.util.Random;
+import java.util.UUID;
+
+/**
+ * 类的功能描述：
+ * 工具类
+ *
+ * @ClassName: ToolsUtil
+ * @Author hcxin
+ * @Date 2017-09-13 03:49:22
+ */
+public class ToolsUtil {
+
+ public synchronized static String getNextCode() {
+ return UUID.randomUUID().toString();
+ }
+
+
+ public synchronized static Float getFee() {
+ Random rand = new Random();
+ float fee = rand.nextFloat();
+
+ return fee;
+ }
+
+
+}

springboot-kafka/src/main/java/com/chen/consumer/MessageConsumer.java

@@ -0,0 +1,31 @@
+package com.chen.consumer;
+
+import com.chen.common.constant.TopicConst;
+import com.chen.common.message.PayMessage;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
+
+/**
+ * 类的功能描述：
+ * 消息消费者者用于处理消息
+ *
+ * @ClassName: MessageConsumer
+ * @Author haichen
+ * @Date 2017-09-13 03:45:55
+ */
+@Service
+public class MessageConsumer {
+ private static Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
+ private Gson gson = new GsonBuilder().create();
+
+ @KafkaListener(topics = TopicConst.PAY_TOPIC)
+ public void onMessage(String payMessage) {
+ PayMessage msg = gson.fromJson(payMessage, PayMessage.class);
+ logger.info("MessageConsumer: onMessage: message is: [" + msg + "]");
+ }
+}

springboot-kafka/src/main/java/com/chen/producer/MessageProducer.java

@@ -0,0 +1,35 @@
+package com.chen.producer;
+
+import com.chen.common.constant.TopicConst;
+import com.chen.common.message.PayMessage;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * 类的功能描述：
+ * 消息生产者用于发送消息
+ *
+ * @ClassName: MessageProducer
+ * @Author hcxin
+ * @Date 2017-09-13 03:45:02
+ */
+@Component
+public class MessageProducer {
+ private static Logger logger = LoggerFactory.getLogger(MessageProducer.class);
+ @Autowired
+ private KafkaTemplate kafkaTemplate;
+ private Gson gson = new GsonBuilder().create();
+
+ public void send(PayMessage payMessage) {
+ String msg = gson.toJson(payMessage);
+ kafkaTemplate.send(TopicConst.PAY_TOPIC, msg);
+ logger.info("MessageProducer: send: message is: [" + msg + "]");
+
+ }
+
+}

springboot-kafka/src/main/resources/application.yml

@@ -0,0 +1,16 @@
+server:
+ port: 8999
+ contextPath : /kafka
+spring:
+ application:
+ name: kafka
+ kafka:
+ bootstrapServers: 192.168.0.195:9092
+ consumer:
+ groupId: myGroup
+ keyDeserializer: org.apache.kafka.common.serialization.StringDeserializer
+ valueDserializer: org.apache.kafka.common.serialization.StringDeserializer
+ producer:
+ groupId: myGroup
+ keyDeserializer: org.apache.kafka.common.serialization.StringSerializer
+ valueDserializer: org.apache.kafka.common.serialization.StringSerializer