部署Kafka+ZK及其日志采集实战(系统版本：linux_CentOs_7.8) ...

• 部署ZK
  1. docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

  复制代码
• 部署Kafka
  1.         docker run -d --name xdclass_kafka \
  2.         -p 9092:9092 \
  3.         -e KAFKA_BROKER_ID=0 \
  4.         --env KAFKA_HEAP_OPTS=-Xmx256M \
  5.         --env KAFKA_HEAP_OPTS=-Xms128M \
  6.         -e KAFKA_ZOOKEEPER_CONNECT=[内网ip]:2181 \
  7.         -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://[外网ip]:9092 \
  8.         -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 wurstmeister/kafka:2.13-2.7.0

  复制代码
• 采用Slf4j采集日志（lombok）
  

• 需求
  
  
  
  • 控制台输出访问日志，方便测试
    
  • 业务数据实际输出到kafka
    
  • 常用的框架 log4j、logback、self4j等
    
  
  
• log4j、logback、self4j 之间有啥关系
  
  
  
  • SLF4J(Simple logging Facade for Java) 门面设计模式 |外观设计模式
    
    
    
    • 把不同的日志系统的实现进行了具体的抽象化，提供统一的日志使用接口
      
    • 具体的日志系统就有log4j,logback等；
      
    • logback也是log4j的作者完成的，有更好的特性，可以取代log4j的一个日志框架, 是slf4j的原生实现
      
    • log4j、logback可以单独的使用，也可以绑定slf4j一起使用
      
    
    
  • 编码规范建议不直接用log4j、logback的API，应该用self4j, 日后更换框架所带来的成本就很低
    
  
  

• 依赖引入
  1. <dependency>
  2.     <groupId>org.projectlombok</groupId>
  3.     <artifactId>lombok</artifactId>
  4. </dependency>
  6. <dependency>
  7.     <groupId>org.springframework.kafka</groupId>
  8.     <artifactId>spring-kafka</artifactId>
  9. </dependency>

  复制代码
• kafka配置application.properties
  1. #----------kafka配置--------------
  2. spring.kafka.bootstrap-servers=[外网ip]:9092
  3. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  4. spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

  复制代码
• logback.xml配置
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <configuration>
  3.     <property name="LOG_HOME" value="./data/logs/link" />
  5.    
  6.     <appender name="console" >
  7.         <encoder>
  8.             <pattern>%date %level [%thread] %logger{36} [%file : %line] %msg%n</pattern>
  9.         </encoder>
  10.     </appender>
  12.    
  13.     <appender name="rollingFile" >
  14.         <file>${LOG_HOME}/link.log</file>
  15.         <rollingPolicy >
  16.             <fileNamePattern>${LOG_HOME}/link-%d{yyyy-MM-dd}.log</fileNamePattern>
  17.         </rollingPolicy>
  18.         <encoder>
  19.             <pattern>%date %level [%thread] %logger{36} [%file : %line] %msg%n</pattern>
  20.         </encoder>
  21.     </appender>
  24.    
  25.     <logger name="net.xdclass.service.impl.LogServiceImpl"
  26.             level="INFO" additivity="false">
  27.         <appender-ref ref="rollingFile" />
  28.         <appender-ref ref="console" />
  29.     </logger>
  31.     <root level="info" additivity="false">
  32.         <appender-ref ref="console" />
  33.     </root>
  35. </configuration>

  复制代码
• LogServiceImpl
  1. @Service
  2. @Slf4j
  3. public class LogServiceImpl implements LogService {
  4.        
  5.         // Kafka：topic
  6.     private static final String TOPIC_NAME = "ods_link_visit_topic";
  8.     @Autowired
  9.     private KafkaTemplate kafkaTemplate;
  11.     /**
  12.      * 记录日志
  13.      *
  14.      * @param request
  15.      * @param shortLinkCode
  16.      * @param accountNo
  17.      * @return
  18.      */
  19.     @Override
  20.     public void recodeShortLinkLog(HttpServletRequest request, String shortLinkCode, Long accountNo) {
  21.         // ip、 浏览器信息
  22.         String ip = CommonUtil.getIpAddr(request);
  23.         // 全部请求头
  24.         Map<String, String> headerMap = CommonUtil.getAllRequestHeader(request);
  26.         Map<String,String> availableMap = new HashMap<>();
  27.         availableMap.put("user-agent",headerMap.get("user-agent"));
  28.         availableMap.put("referer",headerMap.get("referer"));
  29.         availableMap.put("accountNo",accountNo.toString());
  31.         LogRecord logRecord = LogRecord.builder()
  32.                 //日志类型
  33.                 .event(LogTypeEnum.SHORT_LINK_TYPE.name())
  34.                 //日志内容
  35.                 .data(availableMap)
  36.                 //客户端ip
  37.                 .ip(ip)
  38.                 // 时间
  39.                 .ts(CommonUtil.getCurrentTimestamp())
  40.                 //业务唯一标识（短链码）
  41.                 .bizId(shortLinkCode).build();
  43.         String jsonLog = JsonUtil.obj2Json(logRecord);
  45.         //打印日志 in 控制台
  46.         log.info(jsonLog);
  48.         // 发送kafka
  49.         kafkaTemplate.send(TOPIC_NAME,jsonLog);
  51.     }
  52. }

  复制代码
• kafka命令
  1. ```
  2. 创建topic
  3. ./kafka-topics.sh --create --zookeeper 172.17.0.1:2181 --replication-factor 1 --partitions 1 --topic ods_link_visit_topic
  5. 查看topic
  6. ./kafka-topics.sh --list --zookeeper 172.17.0.1:2181
  8. 删除topic
  9. ./kafka-topics.sh --zookeeper 172.17.0.1:2181 --delete --topic ods_link_visit_topic
  11. 消费者消费消息
  12. ./kafka-console-consumer.sh --bootstrap-server 192.168.75.146:9092 --from-beginning --topic ods_link_visit_topic
  14. 生产者发送消息
  15. ./kafka-console-producer.sh --broker-list 192.168.75.146:9092  --topic ods_link_visit_topic
  16. ```

  复制代码
• 测试
  1. @Controller
  2. @Slf4j
  3. public class LinkApiController {
  4.     @Autowired
  5.     private ShortLinkService shortLinkService;
  7.     @Autowired
  8.     private LogService logService;
  11.     /**
  12.      *
  13.      * @param shortLinkCode
  14.      * @param request
  15.      * @param response
  16.      */
  17.     @GetMapping(path = "/test")
  18.     public void dispatch(HttpServletRequest request, HttpServletResponse response) {
  20.         log.info("短链:{}", shortLinkCode);
  21.         logService.recodeShortLinkLog(request, shortLinkCode, shortLinkVO.getAccountNo());
  22.         
  23.     }

  复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！