1.加入kafka传电子路单至数据中台

王通
1 parent 3850fbd3
Showing 3 changed files with 18 additions and 5 deletions
src/main/java/com/bsth/entity/ScheduleRealInfoVo.java
src/main/java/com/bsth/server_rs/schedule/real/ScheduleRealService.java
src/main/resources/application-prod.properties
@@ -154,12 +154,10 @@ public class ScheduleRealInfoVo implements Serializable{
 	/** 关联的公司名称 */
 	private String gsName;
 	/** 关联的公司编码 */
-	@JsonIgnore
 	private String gsBm;
 	/** 关联的分公司名称 */
 	private String fgsName;
 	/** 关联的分公司编码 */
-	@JsonIgnore
 	private String fgsBm;
 	/** 出场顺序号 */
 	@JsonIgnore
@@ -33,8 +33,11 @@ import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
 import org.springframework.messaging.Message;
 import org.springframework.stereotype.Component;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureCallback;
 import javax.ws.rs.*;
 import javax.ws.rs.core.MediaType;
@@ -331,12 +334,24 @@ public class ScheduleRealService implements InitializingBean {
                         ObjectMapper mapper = new ObjectMapper();
                         String json = mapper.writeValueAsString(data);
                         logger.info(json);
-                        kafkaTemplate.send("PDGJ_JQDD", json);
+                        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("PDGJ_JQDD", json);
+                        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
+
+                            @Override
+                            public void onSuccess(SendResult<String, String> result) {
+
+                            }
+
+                            @Override
+                            public void onFailure(Throwable ex) {
+                                logger.error("kafka发送电子路单异常", ex);
+                            }
+                        });
                     }
                 } catch (Exception e) {
                     logger.error("kafka发电子路单调度异常", e);
                 }
             }
-        }, 1, 4, TimeUnit.MINUTES);
+        }, 30, 240, TimeUnit.SECONDS);
     }
 }
@@ -30,7 +30,7 @@ spring.redis.password=bsth_control_001
 spring.redis.port=28008
 #kafka
-spring.kafka.bootstrap-servers=106.54.221.120:9092
+spring.kafka.bootstrap-servers=47.102.42.244:9092
 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
 spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
 spring.kafka.producer.buffer-memory=33554432