应用场景

应用程序集成了多个kafka集群，需根据topic来发送具体的集群；可以抽象为kafka通道，公共kafka通道，http通道（因http是灾备才用）；面对此场景，通过策略模式实现，示例代码如下：

通用接口

  1. package com.yougou.order.kafka.channel;
  3. import com.yougou.order.kafka.model.ProducerModel;
  4. import org.apache.poi.hssf.record.formula.functions.T;
  6. /**
  7.  * 生产者模板-抽象接口
  8.  *
  9.  * @author david
  10.  * @site www.belle.net.cn
  11.  * @company 百丽国际
  12.  * @Date 创建时间：2021-04-13 21:36
  13.  */
  14. public interface IProducerChannel {
  16.     /**
  17.      * 抽象接口，发送数据
  18.      *
  19.      * @param dto
  20.      * @return
  21.      * @throws Exception
  22.      */
  23.     boolean sendData(ProducerModel dto) throws Exception;
  24. }
  1. package com.yougou.order.kafka.channel;
  3. import co.elastic.logging.ComplexLogGenerator;
  4. import com.alibaba.fastjson.JSON;
  5. import com.google.gson.Gson;
  6. import com.yougou.framework.logger.Logger;
  7. import com.yougou.framework.logger.LoggerFactory;
  8. import com.yougou.order.kafka.DsKafkaProducerServer;
  9. import com.yougou.order.kafka.KafkaMesConstant;
  10. import com.yougou.order.kafka.dto.KafkaDto;
  11. import com.yougou.order.kafka.model.ProducerModel;
  12. import com.yougou.tools.common.utils.UUIDGenerator;
  13. import org.apache.commons.lang3.StringUtils;
  14. import org.apache.poi.hssf.record.formula.functions.T;
  15. import org.springframework.beans.factory.annotation.Autowired;
  16. import org.springframework.beans.factory.annotation.Qualifier;
  17. import org.springframework.kafka.core.KafkaTemplate;
  18. import org.springframework.kafka.support.SendResult;
  19. import org.springframework.stereotype.Service;
  20. import org.springframework.util.concurrent.ListenableFuture;
  22. import java.util.HashMap;
  23. import java.util.Map;
  25. /**
  26.  * 公共kafka集群-生产者模板
  27.  *
  28.  * @author david
  29.  * @site www.belle.net.cn
  30.  * @company 百丽国际
  31.  * @Date 创建时间：2021-04-13 21:36
  32.  */
  33. @Service
  34. public class PubKafkaChannel implements IProducerChannel {
  36.     private final Logger logger = LoggerFactory.getLogger(this.getClass());
  38.     @Autowired
  39.     @Qualifier("pubKafkaTemplate")
  40.     private KafkaTemplate<String, String> pubKafkaTemplate;
  42.     @Override
  43.     public boolean sendData(ProducerModel dto) {
  44.         try {
  45.             if (StringUtils.isBlank(dto.getProducerEnum().getTopic())) {
  46.                 logger.error(dto.getProducerEnum().getBizName() + "未配置topic");
  47.                 return false;
  48.             }
  49.             String msgId = UUIDGenerator.getUUID();
  50.             KafkaDto kafkaDto = new KafkaDto();
  51.             kafkaDto.setData(dto);
  52.             kafkaDto.setMsgId(msgId);
  53.             String reqJsonStr = JSON.toJSONString(kafkaDto);
  54.             // 调用底层kafka接口
  55.             logger.info("发送[pub]" + dto.getProducerEnum().getTopic() + "数据:" + reqJsonStr);
  56.             pubKafkaTemplate.send(dto.getProducerEnum().getTopic(), reqJsonStr);
  57.             return true;
  58.         } catch (Exception e) {
  59.             logger.error(dto.getProducerEnum().getBizName() + "发送异常{}", e.getMessage());
  60.             return false;
  61.         }
  62.     }
  63. }
  1. package com.yougou.order.kafka.channel;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.yougou.framework.logger.Logger;
  6. import com.yougou.framework.logger.LoggerFactory;
  7. import com.yougou.order.kafka.dto.KafkaDto;
  8. import com.yougou.order.kafka.model.ProducerModel;
  9. import com.yougou.ordercenter.common.PropertiesUtil;
  10. import com.yougou.ordercenter.utils.HttpClient;
  11. import com.yougou.tools.common.utils.UUIDGenerator;
  12. import org.apache.commons.lang3.StringUtils;
  13. import org.apache.poi.hssf.record.formula.functions.T;
  14. import org.springframework.stereotype.Service;
  16. import java.util.HashMap;
  17. import java.util.Map;
  19. /**
  20.  * http-生产者模板
  21.  *
  22.  * @author david
  23.  * @site www.belle.net.cn
  24.  * @company 百丽国际
  25.  * @Date 创建时间：2021-04-13 21:36
  26.  */
  27. @Service
  28. public class HttpChannel implements IProducerChannel {
  30.     private final Logger logger = LoggerFactory.getLogger(this.getClass());
  32.     @Override
  33.     public boolean sendData(ProducerModel dto) {
  34.         try {
  35.             if (StringUtils.isBlank(dto.getProducerEnum().getUrl())) {
  36.                 logger.error(dto.getProducerEnum().getBizName() + "未配置url");
  37.                 return false;
  38.             }
  39.             String fullUrl = PropertiesUtil.getInstance().getIspHost() + dto.getProducerEnum().getUrl();
  40.             String msgId = UUIDGenerator.getUUID();
  41.             KafkaDto kafkaDto = new KafkaDto();
  42.             kafkaDto.setData(dto);
  43.             kafkaDto.setMsgId(msgId);
  44.             String reqJsonStr = JSON.toJSONString(kafkaDto);
  45.             // 调用底层kafka接口
  46.             logger.info("发送[http]" + fullUrl + "数据:" + reqJsonStr);
  47.             return sendByHttp(fullUrl, reqJsonStr);
  48.         } catch (Exception e) {
  49.             logger.error(dto.getProducerEnum().getBizName() + "发送异常{}", e.getMessage());
  50.             return false;
  51.         }
  52.     }
  54.     /**
  55.      * 发送isp底层http接口
  56.      *
  57.      * @param url
  58.      * @param httpJson
  59.      * @return
  60.      * @throws Exception
  61.      */
  62.     public boolean sendByHttp(String url, String httpJson) throws Exception {
  63.         // 请求报文组包
  64.         Map<String, String> reqParamMap = new HashMap<String, String>();
  65.         // 组装发送到isp的报文
  66.         reqParamMap.put("appKey", PropertiesUtil.getInstance().getIspKey());
  67.         reqParamMap.put("orderDetail", httpJson);
  69.         HttpClient httpClient = new HttpClient(url, 30000, 30000, true);
  70.         String rev = httpClient.send(reqParamMap);
  72.         JSONObject resultObj = JSON.parseObject(rev);
  73.         if (resultObj != null && org.apache.commons.lang.StringUtils.equals(resultObj.getString("code"), "200")) {
  74.             return true;
  75.         } else {
  76.             return false;
  77.         }
  78.     }
  80. }

具体调用方，声明用接口，然后用接口的子类来实现；