部署Kafka+ZK及其日誌採集實戰(系統版本：linux_CentOs_7.8)

部署ZK

docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

部署Kafka

	docker run -d --name xdclass_kafka \
	-p 9092:9092 \
	-e KAFKA_BROKER_ID=0 \
	--env KAFKA_HEAP_OPTS=-Xmx256M \
	--env KAFKA_HEAP_OPTS=-Xms128M \
	-e KAFKA_ZOOKEEPER_CONNECT=[內網ip]:2181 \
	-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://[外網ip]:9092 \
	-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, 日後更換框架所帶來的成本就很低

依賴引入

<!-- 代碼自動生成依賴 end-->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>
<!--Springboot項目整合spring-kafka依賴包配置-->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

kafka配置application.properties

#----------kafka配置--------------
spring.kafka.bootstrap-servers=[外網ip]:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

logback.xml配置

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <property name="LOG_HOME" value="./data/logs/link" />

    <!--採用列印到控制台，記錄日誌的方式-->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%date %level [%thread] %logger{36} [%file : %line] %msg%n</pattern>
        </encoder>
    </appender>

    <!-- 採用保存到日誌文件 記錄日誌的方式-->
    <appender name="rollingFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_HOME}/link.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${LOG_HOME}/link-%d{yyyy-MM-dd}.log</fileNamePattern>
        </rollingPolicy>
        <encoder>
            <pattern>%date %level [%thread] %logger{36} [%file : %line] %msg%n</pattern>
        </encoder>
    </appender>


    <!-- 指定某個類單獨列印日誌 -->
    <logger name="net.xdclass.service.impl.LogServiceImpl"
            level="INFO" additivity="false">
        <appender-ref ref="rollingFile" />
        <appender-ref ref="console" />
    </logger>

    <root level="info" additivity="false">
        <appender-ref ref="console" />
    </root>

</configuration>

LogServiceImpl

@Service
@Slf4j
public class LogServiceImpl implements LogService {
	
	// Kafka：topic
    private static final String TOPIC_NAME = "ods_link_visit_topic";

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 記錄日誌
     *
     * @param request
     * @param shortLinkCode
     * @param accountNo
     * @return
     */
    @Override
    public void recodeShortLinkLog(HttpServletRequest request, String shortLinkCode, Long accountNo) {
        // ip、 瀏覽器信息
        String ip = CommonUtil.getIpAddr(request);
        // 全部請求頭
        Map<String, String> headerMap = CommonUtil.getAllRequestHeader(request);

        Map<String,String> availableMap = new HashMap<>();
        availableMap.put("user-agent",headerMap.get("user-agent"));
        availableMap.put("referer",headerMap.get("referer"));
        availableMap.put("accountNo",accountNo.toString());

        LogRecord logRecord = LogRecord.builder()
                //日誌類型
                .event(LogTypeEnum.SHORT_LINK_TYPE.name())
                //日誌內容
                .data(availableMap)
                //客戶端ip
                .ip(ip)
                // 時間
                .ts(CommonUtil.getCurrentTimestamp())
                //業務唯一標識（短鏈碼）
                .bizId(shortLinkCode).build();

        String jsonLog = JsonUtil.obj2Json(logRecord);

        //列印日誌 in 控制台
        log.info(jsonLog);

        // 發送kafka
        kafkaTemplate.send(TOPIC_NAME,jsonLog);

    }
}

kafka命令

```
創建topic
./kafka-topics.sh --create --zookeeper 172.17.0.1:2181 --replication-factor 1 --partitions 1 --topic ods_link_visit_topic

查看topic
./kafka-topics.sh --list --zookeeper 172.17.0.1:2181

刪除topic
./kafka-topics.sh --zookeeper 172.17.0.1:2181 --delete --topic ods_link_visit_topic

消費者消費消息
./kafka-console-consumer.sh --bootstrap-server 192.168.75.146:9092 --from-beginning --topic ods_link_visit_topic

生產者發送消息
./kafka-console-producer.sh --broker-list 192.168.75.146:9092  --topic ods_link_visit_topic
```

測試

@Controller
@Slf4j
public class LinkApiController {
    @Autowired
    private ShortLinkService shortLinkService;

    @Autowired
    private LogService logService;


    @GetMapping(path = "/test")
    public void dispatch(HttpServletRequest request, HttpServletResponse response) {

        logService.recodeShortLinkLog(request, shortLinkCode, shortLinkVO.getAccountNo());
        
    }