• Spring 框架下如何调用kafka

    1、Spring 项目代码结构如下：

    2、数据库资源配置文件如下：

    #sql配置文件
spring.datasource.driver-class-name=com.microsoft.sqlserver.jdbc.SQLServerDriver
#.19為測試地址，.13為正式地址
spring.datasource.url=jdbc:sqlserver://172.12.100.19:1433;DatabaseName=data
spring.datasource.username=wms1234
spring.datasource.password=wms1234
spring.datasource.hikari.connection-timeout=20000
spring.datasource.hikari.connection-test-query=SELECT 1 
#mybatis配置
mybatis.mapper-locations=classpath:mapper/*.xml
#配置映射的实体类
mybatis.type-aliases-package=com.device.search.dao

    3、Kafaka实现类源码 

    1. package com.device.search.kafka;
    2.  
    3. import java.io.File;
    4. import java.io.IOException;
    5. import java.security.Timestamp;
    6. import java.util.ArrayList;
    7. import java.util.Date;
    8. import java.util.List;
    9. import java.util.Properties;
    10.  
    11. import com.aliyun.api.internal.mapping.Converter;
    12. import com.aliyun.api.internal.parser.json.JsonConverter;
    13. import com.device.search.dao.AimerBoxlistDao;
    14. import com.device.search.dao.AimerPackboxDao;
    15. import com.device.search.dao.AimerToWms2BDao;
    16. import com.device.search.model.EdiOutErpData;
    17. import com.device.search.model.PackListData;
    18. import com.device.search.service.AimerBoxlistService;
    19. import com.device.search.service.AimerPackboxService;
    20. import com.device.search.service.AimerToWms2BService;
    21. import com.device.search.utils.TimeUtil;
    22. import com.fasterxml.jackson.databind.JsonSerializer;
    23. import com.fasterxml.jackson.databind.annotation.JsonSerialize;
    24. import com.google.gson.Gson;
    25. import com.qimen.api.request.DeliveryorderBatchcreateRequest;
    26. import org.apache.kafka.clients.producer.Callback;
    27. import org.apache.kafka.clients.producer.KafkaProducer;
    28. import org.apache.kafka.clients.producer.ProducerRecord;
    29. import org.apache.kafka.clients.producer.RecordMetadata;
    30. import org.slf4j.Logger;
    31. import org.slf4j.LoggerFactory;
    32. import com.device.search.model.SendData;
    33. public class PackListSendThread implements Runnable {
    34.  
    35.   private static Logger logger = LoggerFactory.getLogger(PackListSendThread.class);
    36.   private final KafkaProducer producer;
    37.   private final String topic;
    38.   public static String packListBack_flag="0";
    39.   private AimerPackboxService aimerPackboxService;
    40.   private AimerBoxlistService aimerBoxlistService;
    41.   private AimerToWms2BService aimerToWms2BService;
    42.   private  final String log_path="d:\\log\\PackListBack\\";
    43.   private  final String err_log="PackListBack_err_";
    44.   private  final String rsp_log="PackListBack_rsp_";
    45.   private  final String req_log="PackListBack_req_";
    46.   public PackListSendThread(String brokers, String topic, AimerPackboxService aimerPackboxService, AimerBoxlistService aimerBoxlistService, AimerToWms2BService aimerToWms2BService)
    47.   {
    48.     Properties prop = createProducerConfig(brokers);
    49.     this.producer = new KafkaProducer(prop);
    50.     this.topic = topic;
    51.     this.aimerBoxlistService=aimerBoxlistService;
    52.     this.aimerPackboxService=aimerPackboxService;
    53.     this.aimerToWms2BService=aimerToWms2BService;
    54.   }
    55.  
    56.   private static Properties createProducerConfig(String brokers) {
    57.     Properties props = new Properties();
    58.     props.put("bootstrap.servers", brokers);
    59.     props.put("acks", "all");
    60.     props.put("retries", 0);
    61.     props.put("batch.size", 16384);
    62.     props.put("linger.ms", 1);
    63.     props.put("buffer.memory", 33554432);
    64.     props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    65.     props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    66.     return props;
    67.   }
    68.  
    69.   @Override
    70.   public void run() {
    71.     System.out.println("Produces 3 messages");
    72.     while (true) {
    73.        try {
    74. //         if (packListBack_flag == "0") {
    75. //           continue;
    76. //         }
    77.          synchronized (this) {
    78.            if (packListBack_flag == "0") {
    79.              continue;
    80.            }
    81.          }
    82.          List listPackbox = aimerPackboxService.findbyid("N");
    83.          for (int i = 0; i < listPackbox.size(); i++)
    84.          {
    85.            String finalBox_no = listPackbox.get(i).getBoxCode();
    86.            String sotype="SC";
    87.            PackListData packListData = new PackListData();
    88.  
    89.            packListData.setWarehouseID(listPackbox.get(i).getWhsId());
    90.            packListData.setCARTONID(finalBox_no);
    91.            packListData.setConsigneeid(listPackbox.get(i).getCustom());
    92.            packListData.setWaveno(listPackbox.get(i).getTicno());
    93.            packListData.setGrossweight(listPackbox.get(i).getWeight());
    94.            packListData.setCartontype(listPackbox.get(i).getBoxType());
    95.            packListData.setPackno(listPackbox.get(i).getPackPort());
    96.            packListData.setPackwho(listPackbox.get(i).getOperator());
    97.  
    98.            packListData.setStarttime(listPackbox.get(i).getCreateDate());
    99.            packListData.setEndtime(listPackbox.get(i).getDoneDate()); //20220428
    100.            //2022.04.08 toC箱单增加运单号和承运商
    101.            packListData.setCarrierid(listPackbox.get(i).getCysname());
    102.            packListData.setDeliverno(listPackbox.get(i).getMailno());
    103.  
    104.            List listBoxlist = aimerBoxlistService.findbyboxno(finalBox_no);
    105.            long shipqty=0;
    106.            if (listBoxlist.size()<=0)//没有箱明细报错
    107.            {
    108.              logger.error("回传箱明细：没有箱明细数据"+finalBox_no);
    109.              aimerPackboxService.update_boxno_err(finalBox_no,finalBox_no+"回传箱明细：没有箱明细数据");
    110.              continue;
    111.            }
    112.            String ordno=listBoxlist.get(0).getOrdno();
    113.            List listAimerToWms2BDao = aimerToWms2BService.find2bOrdnoUsr02(ordno,"2");
    114.            if (listAimerToWms2BDao.size()<=0)//没有listAimerToWms2BDao
    115.            {
    116.              logger.error("Aimer_to_wms明细：没有出库数据"+finalBox_no);
    117.              aimerPackboxService.update_boxno_err(finalBox_no,finalBox_no+":"+ordno+"Aimer_to_wms明细：没有出库数据");
    118.              continue;
    119.            }
    120.  
    121.            List listReturnBoxlist = new ArrayList();
    122.            for (AimerBoxlistDao aimerBoxlistDao :listBoxlist)
    123.            {
    124.              //20220402  循环查询箱号对应每个单据的操作号,有一个箱号里多个单子。
    125.              List aimerToWms2BDaoList = aimerToWms2BService.find2bOrdnoUsr02(aimerBoxlistDao.getOrdno(),"2");
    126.              if (aimerToWms2BDaoList.size()<=0)//没有listAimerToWms2BDao
    127.              {
    128.                logger.error("Aimer_to_wms明细：没有出库数据11"+finalBox_no);
    129.                aimerPackboxService.update_boxno_err(finalBox_no,finalBox_no+":"+aimerBoxlistDao.getOrdno()+"Aimer_to_wms明细1：没有出库数据");
    130.                continue;
    131.              }
    132.               shipqty = (long) (shipqty +  Double.parseDouble(aimerBoxlistDao.getBoxQty()));
    133.               EdiOutErpData ediOutErpData = new EdiOutErpData();
    134.               ediOutErpData.setSku(aimerBoxlistDao.getBarCode());
    135.               ediOutErpData.setQty(aimerBoxlistDao.getBoxQty());
    136.               ediOutErpData.setSoreference1(aimerToWms2BDaoList.get(0).getUsr02());
    137.               //toC出库箱单明细操作好通过KAFKA回传 --2022.04.08
    138.                if(listAimerToWms2BDao.get(0).getOtype().equals("全渠道O2O") || listAimerToWms2BDao.get(0).getOtype().equals("全渠道o2o")){
    139.                  ediOutErpData.setSoreference1(aimerToWms2BDaoList.get(0).getOrdno());
    140.                }
    141.               ediOutErpData.setNotes(aimerToWms2BDaoList.get(0).getOrdno());
    142.               listReturnBoxlist.add(ediOutErpData);
    143.            }
    144.            if (listAimerToWms2BDao.get(0).getUsr01()!=null)//202203029
    145.            {
    146.              if (listAimerToWms2BDao.get(0).getUsr01().equals("B2BCK"))
    147.              {
    148.                sotype = "SC";
    149.              }
    150.              if (listAimerToWms2BDao.get(0).getUsr01().equals("DBCK"))
    151.              {
    152.                sotype = "SC";
    153.              }
    154.              if (listAimerToWms2BDao.get(0).getUsr01().equals("SQDCK"))
    155.              {
    156.                sotype = "OT";
    157.              }
    158.            }
    159.            //toC出库箱单明细通过KAFKA回传 --2022.04.08
    160.            if(listAimerToWms2BDao.get(0).getOtype().equals("全渠道O2O") || listAimerToWms2BDao.get(0).getOtype().equals("全渠道o2o")){
    161.              sotype = "SO";
    162.            }
    163.            packListData.setSOType(sotype);
    164.            packListData.setShippedQty(shipqty);
    165.            packListData.setItemData(listReturnBoxlist);
    166.            SendData sendData = new SendData();
    167.            sendData.setopr_type("add");
    168.            long time = System.currentTimeMillis();
    169.            sendData.settimestamp(Long.toString(time));
    170.            sendData.setPackListData(packListData);
    171.  
    172.            String msg = new Gson().toJson(sendData);
    173.  
    174.            File file=new File(log_path+ TimeUtil.convertDay(new Date()));
    175.            if(!file.exists()){//如果文件夹不存在
    176.              file.mkdirs();//创建文件夹
    177.            }
    178.            String req_filename=log_path+TimeUtil.convertDay(new Date())+"\\"+ req_log+packListData.getCARTONID() +"_" +TimeUtil.convertDate(new Date())+".xml" ;
    179.            TimeUtil.OutJson(msg,req_filename);
    180.  
    181.            producer.send(new ProducerRecord(topic, msg), new Callback()
    182.            {
    183.              public void onCompletion(RecordMetadata metadata, Exception e)
    184.              {
    185.                String resString = new Gson().toJson(metadata);
    186.                if (e != null)
    187.                {
    188.                  String req_filename=log_path+TimeUtil.convertDay(new Date())+"\\"+ err_log+packListData.getCARTONID() +"_" +TimeUtil.convertDate(new Date())+".xml" ;
    189.                  try {
    190.                    TimeUtil.OutJson(resString,req_filename);
    191.                  } catch (IOException ioException) {
    192.                    ioException.printStackTrace();
    193.                  }
    194.                  e.printStackTrace();
    195.                }
    196.                String req_filename=log_path+TimeUtil.convertDay(new Date())+"\\"+ rsp_log+packListData.getCARTONID() +"_" +TimeUtil.convertDate(new Date())+".xml" ;
    197.                try {
    198.                  TimeUtil.OutJson(resString,req_filename);
    199.                } catch (IOException ioException) {
    200.                  ioException.printStackTrace();
    201.                }
    202.                System.out.println("Sent:" + msg + ", Partition: " + metadata.partition() + ", Offset: "
    203.                        + metadata.offset());
    204.                aimerPackboxService.update_boxno(finalBox_no);
    205.              }
    206.            });
    207.          }
    208.          // closes producer
    209.          //producer.close();
    210.          Thread.sleep(3000);
    211.        }catch (Exception ex)
    212.        {
    213.          String req_filename=log_path+TimeUtil.convertDay(new Date())+"\\"+ rsp_log+"eRR_" +TimeUtil.convertDate(new Date())+".xml" ;
    214.          try {
    215.            TimeUtil.OutJson(ex.getMessage(),req_filename);
    216.          } catch (IOException e) {
    217.            e.printStackTrace();
    218.          }
    219.          continue;
    220.        }
    221.     }
    222.   }
    223. }

  • 相关阅读:
    if消除术之 Map + Function
    DevExpress开发WPF应用实现对话框总结
    SpringCloud和SpringBoot版本对应关系
    【C++进阶】：C++类型转换
    UE4 Sequence添加基础动画效果 （04-在序列中使用粒子效果）
    《IEEE Transactions on Robotics》发表！北京大学研究团队推出具有多种运动模态的软体两栖机器人
    SpringBoot整合knife 4j
    基于SOCKET编程多人聊天软件HTML+JSON报文
    什么是集成测试？集成测试方法有哪些？
    ChatGPT突然上线APP！iPhone可用、速度更快，GPT-4用量限制疑似取消
  • 原文地址：https://blog.csdn.net/xdpcxq/article/details/128182638