update: 示例代码

Author: Wizzercn

mqtt-zoo/mqtt-test-kafka/pom.xml

@@ -10,11 +10,12 @@
     <properties>
         <mqttwk.version>1.3.0-netty</mqttwk.version>
         <nutzboot.version>2.4.1.v20201014</nutzboot.version>
-        <netty.version>4.1.28.Final</netty.version>
-        <fastjson.version>1.2.55</fastjson.version>
-        <hutool.version>4.1.2</hutool.version>
+        <netty.version>4.1.77.Final</netty.version>
+        <fastjson.version>2.0.4</fastjson.version>
+        <hutool.version>5.8.1</hutool.version>
         <kafka_2.12.version>2.0.0</kafka_2.12.version>
-        <slf4j-log4j12.version>1.7.25</slf4j-log4j12.version>
+        <slf4j.version>1.7.25</slf4j.version>
+        <logback.version>1.2.9</logback.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     </properties>
@@ -56,10 +57,21 @@
             <artifactId>kafka_2.12</artifactId>
             <version>${kafka_2.12.version}</version>
         </dependency>
+        <!-- logback -->
         <dependency>
             <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-log4j12</artifactId>
-            <version>${slf4j-log4j12.version}</version>
+            <artifactId>slf4j-api</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-core</artifactId>
+            <version>${logback.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <version>${logback.version}</version>
         </dependency>
     </dependencies>
     <repositories>

mqtt-zoo/mqtt-test-kafka/src/main/resources/META-INF/nutz/org.nutz.boot.config.ConfigureLoader

@@ -0,0 +1 @@
+org.nutz.boot.config.impl.YamlConfigureLoader

mqtt-zoo/mqtt-test-kafka/src/main/resources/application.properties

@@ -0,0 +1,40 @@
+nutz:
+  application:
+    name: test
+
+mqttwk:
+  broker:
+    kafka:
+      # 是否启用kafka消息转发
+      broker-enabled: false
+      bootstrap:
+        # kafka地址 127.0.0.1:9092,127.0.0.1:9093
+        servers: 127.0.0.1:9092
+      # acks回令 如果必须等待回令,那么设置acks为all,否则，设置为-1,等待回令会有性能损耗
+      acks: -1
+      # 重试次数
+      retries: 3
+      batch:
+        # 批量提交大小
+        size: 16384
+      linger:
+        # 提交延迟等待时间（等待时间内可以追加提交）
+        ms: 1
+      buffer:
+        # 缓存大小
+        memory: 33554432
+      key:
+        # 序列化方式
+        serializer: org.apache.kafka.common.serialization.StringSerializer
+      value:
+        # 序列化方式
+        serializer: org.apache.kafka.common.serialization.StringSerializer
+      partitioner:
+        class: cn.wizzer.iot.mqtt.server.store.kafka.SimplePartitioner
+      producer:
+        # kafka转发的主题
+        topic: mqtt_publish
+      consumer:
+        topic: mqtt_subscribe
+      group:
+        id: mqttwk

mqtt-zoo/mqtt-test-kafka/src/main/resources/application.yaml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<configuration scan="false" scanPeriod="60000" debug="false">
+
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <layout class="ch.qos.logback.classic.PatternLayout">
+            <pattern>%highlight(%date{yyyy-MM-dd HH:mm:ss.SSS}) %cyan([%thread]) %yellow(%-5level) %green(%logger{36}).%gray(%M)-%boldMagenta(%line) - %blue(%msg%n)</pattern>
+        </layout>
+    </appender>
+
+    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+    <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+      <!-- rollover daily -->
+      <fileNamePattern>logs/test-%d{yyyy-MM-dd}.%i.txt</fileNamePattern>
+       <!-- each file should be at most 100MB, keep 60 days worth of history, but at most 20GB -->
+       <maxFileSize>50MB</maxFileSize>
+       <maxHistory>30</maxHistory>
+       <totalSizeCap>1GB</totalSizeCap>
+    </rollingPolicy>
+    <encoder>
+      <pattern>[%-5level] %d{HH:mm:ss.SSS} %logger - %msg%n</pattern>
+    </encoder>
+  </appender>
+
+
+    <!-- 异步输出 -->
+    <appender name ="ASYNC" class= "ch.qos.logback.classic.AsyncAppender">
+        <!-- 不丢失日志.默认的,如果队列的80%已满,则会丢弃TRACT、DEBUG、INFO级别的日志 -->
+        <discardingThreshold>0</discardingThreshold>
+        <!-- 更改默认的队列的深度,该值会影响性能.默认值为256 -->
+        <queueSize>256</queueSize>
+        <!-- 添加附加的appender,最多只能添加一个 -->
+        <appender-ref ref ="FILE"/>
+    </appender>
+
+    <logger name="java" additivity="false" />
+    <logger name="org.eclipse.jetty" level="INFO">
+        <appender-ref ref="STDOUT" />
+    </logger>
+    <root level="DEBUG">
+        <appender-ref ref="STDOUT" />
+        <appender-ref ref="ASYNC" />
+    </root>
+
+</configuration>