kafka test go!

pipeline not done

austin-service-api-impl/pom.xml

@@ -11,5 +11,11 @@
 
     <artifactId>austin-service-api-impl</artifactId>
 
-
+    <dependencies>
+        <dependency>
+            <groupId>com.java3y.austin</groupId>
+            <artifactId>austin-service-api</artifactId>
+            <version>0.0.1-SNAPSHOT</version>
+        </dependency>
+    </dependencies>
 </project>

austin-service-api-impl/src/main/java/com/java3y/austin/pipeline/BusinessProcess.java

@@ -0,0 +1,10 @@
+package com.java3y.austin.pipeline;
+
+/**
+ * 业务执行器
+ *
+ * @author 3y
+ */
+public interface BusinessProcess {
+    void process(ProcessContext context);
+}

austin-service-api-impl/src/main/java/com/java3y/austin/pipeline/ProcessContext.java

@@ -0,0 +1,24 @@
+package com.java3y.austin.pipeline;
+
+/**
+ * 责任链上下文
+ * @author 3y
+ */
+public class ProcessContext {
+
+    /**
+     * 标识责任链的code
+     */
+    private String code;
+
+    /**
+     * 存储责任链上下文数据的模型
+     */
+    private ProcessModel processModel;
+
+    /**
+     * 责任链中断的标识
+     */
+    private Boolean needBreak = false;
+
+}

austin-service-api-impl/src/main/java/com/java3y/austin/pipeline/ProcessModel.java

@@ -0,0 +1,10 @@
+package com.java3y.austin.pipeline;
+
+
+/**
+ * 真正存储着责任链的数据
+ * @author 3y
+ */
+public class ProcessModel {
+
+}

austin-service-api-impl/src/main/java/com/java3y/austin/pipeline/ProcessTemplate.java

@@ -0,0 +1,19 @@
+package com.java3y.austin.pipeline;
+
+import java.util.List;
+
+/**
+ * 业务执行模板（把责任链的逻辑串起来）
+ * @author 3y
+ */
+public class ProcessTemplate {
+
+    private List<BusinessProcess> processList;
+
+    public List<BusinessProcess> getProcessList() {
+        return processList;
+    }
+    public void setProcessList(List<BusinessProcess> processList) {
+        this.processList = processList;
+    }
+}

austin-service-api-impl/src/main/java/com/java3y/austin/service/SendServiceImpl.java

@@ -0,0 +1,20 @@
+package com.java3y.austin.service;
+
+import com.java3y.austin.domain.SendRequest;
+import com.java3y.austin.domain.SendResponse;
+
+/**
+ * 发送接口
+ * @author 3y
+ */
+public class SendServiceImpl implements SendService  {
+    @Override
+    public SendResponse send(SendRequest sendRequest) {
+        return null;
+    }
+
+    @Override
+    public SendResponse batchSend(SendRequest sendRequest) {
+        return null;
+    }
+}

austin-service-api/src/main/java/com/java3y/austin/domain/SendRequest.java

@@ -0,0 +1,23 @@
+package com.java3y.austin.domain;
+
+
+/**
+ * 发送接口的参数
+ *
+ */
+public class SendRequest {
+
+    /**
+     * 消息模板Id
+     */
+    private Long messageTemplateId;
+
+    /**
+     * 接收者
+     * 多个用
+     */
+    private String receiver;
+
+
+
+}

austin-service-api/src/main/java/com/java3y/austin/domain/SendResponse.java

@@ -0,0 +1,4 @@
+package com.java3y.austin.domain;
+
+public class SendResponse {
+}

austin-service-api/src/main/java/com/java3y/austin/service/SendService.java

@@ -0,0 +1,18 @@
+package com.java3y.austin.service;
+
+import com.java3y.austin.domain.SendRequest;
+import com.java3y.austin.domain.SendResponse;
+
+/**
+ * 发送接口
+ *
+ * @author 3y
+ */
+public interface SendService {
+
+
+    SendResponse send(SendRequest sendRequest);
+
+
+    SendResponse batchSend(SendRequest sendRequest);
+}

austin-web/pom.xml

@@ -21,6 +21,10 @@
             <artifactId>austin-handler</artifactId>
             <version>0.0.1-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

austin-web/src/main/java/com/java3y/austin/controller/KafkaTestController.java

@@ -0,0 +1,24 @@
+package com.java3y.austin.controller;
+
+import com.java3y.austin.kafkatest.UserLogProducer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+public class KafkaTestController {
+
+    @Autowired
+    private UserLogProducer userLogProducer;
+
+    /**
+     * test insert
+     */
+    @GetMapping("/kafka/insert")
+    public String insert(String userId) {
+        userLogProducer.sendLog(userId);
+
+        return null;
+    }
+
+}

austin-web/src/main/java/com/java3y/austin/kafkatest/UserLog.java

@@ -0,0 +1,18 @@
+package com.java3y.austin.kafkatest;
+
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+/**
+ * @Author 18011618
+ * @Description 定义用户发送的日志数据
+ * @Date 14:42 2018/7/20
+ * @Modify By
+ */
+@Data
+@Accessors(chain = true)
+public class UserLog {
+    private String username;
+    private String userid;
+    private String state;
+}

austin-web/src/main/java/com/java3y/austin/kafkatest/UserLogConsumer.java

@@ -0,0 +1,30 @@
+package com.java3y.austin.kafkatest;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.Optional;
+
+/**
+ * @Author 18011618
+ * @Description
+ * @Date 14:50 2018/7/20
+ * @Modify By
+ */
+@Component
+@Slf4j
+public class UserLogConsumer {
+    @KafkaListener(topics = {"austin"},groupId = "austinGroup1")
+    public void consumer(ConsumerRecord<?,?> consumerRecord){
+        //判断是否为null
+        Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
+        log.info(">>>>>>>>>> record =" + kafkaMessage);
+        if(kafkaMessage.isPresent()){
+            //得到Optional实例中的值
+            Object message = kafkaMessage.get();
+            System.err.println("消费消息:"+message);
+        }
+
+    }
+}

austin-web/src/main/java/com/java3y/austin/kafkatest/UserLogProducer.java

@@ -0,0 +1,29 @@
+package com.java3y.austin.kafkatest;
+
+import com.alibaba.fastjson.JSON;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Author 18011618
+ * @Description
+ * @Date 14:43 2018/7/20
+ * @Modify By
+ */
+@Component
+public class UserLogProducer {
+    @Autowired
+    private KafkaTemplate kafkaTemplate;
+
+    /**
+     * 发送数据
+     * @param userid
+     */
+    public void sendLog(String userid){
+        UserLog userLog = new UserLog();
+        userLog.setUsername("jhp").setUserid(userid).setState("0");
+        System.err.println("发送用户日志数据:"+userLog);
+        kafkaTemplate.send("austin", JSON.toJSONString(userLog));
+    }
+}

austin-web/src/main/resources/application.yml

@@ -25,5 +25,17 @@ spring:
     username:
     password:
     driver-class-name:
+  # kafka相关的信息配置 TODO
+  kafka:
+    bootstrap-servers:
+    producer:
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
+    consumer:
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      auto:
+        offset:
+          reset: earliest
 
 # tomcat / HikariPool(数据库连接池 配置） TODO