完善文档和示例代码

Author: codingmiao

.idea/encodings.xml

@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.wowtools.hppt</groupId>
+        <artifactId>hppt</artifactId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>kafkademo</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.wowtools.hppt</groupId>
+            <artifactId>run</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>3.4.0</version>
+        </dependency>
+    </dependencies>
+
+</project>

_doc/img/3.jpg

@@ -0,0 +1,60 @@
+package org.wowtools.hppt.kafkademo;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.wowtools.hppt.run.sc.common.ClientSessionService;
+import org.wowtools.hppt.run.sc.pojo.ScConfig;
+import org.wowtools.hppt.run.ss.pojo.SsConfig;
+
+import java.util.ArrayList;
+import java.util.Properties;
+
+/**
+ * 客户端，部署在电脑A上
+ * @author liuyu
+ * @date 2024/6/15
+ */
+public class ClientDemo extends ClientSessionService {
+    //TODO 传输文件等大字节数传播的情况下，需处理kafka字节顺序消费问题
+    public ClientDemo(ScConfig config) throws Exception {
+        super(config);
+    }
+
+    private KafkaUtil.BytesFunction sendToServer;
+    private KafkaUtil.BytesFunction clientConsumer;
+    @Override
+    protected void connectToServer(ScConfig config, Cb cb) throws Exception {
+        //初始化时构造好向kafka生产和消费数据的工具
+        sendToServer = KafkaUtil.buildProducer(KafkaUtil.ClientSendTopic);
+
+        clientConsumer = (bytes) -> {
+            //消费到客户端的数据，调用receiveServerBytes方法来接收
+            try {
+                receiveServerBytes(bytes);
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        };
+        KafkaUtil.buildConsumer("client", KafkaUtil.ServerSendTopic, clientConsumer);
+        cb.end();//调用end方法，通知框架连接完成
+    }
+
+    @Override
+    protected void sendBytesToServer(byte[] bytes) {
+        sendToServer.f(bytes);
+    }
+
+    public static void main(String[] args) throws Exception{
+        ScConfig cfg = new ScConfig();
+        cfg.clientId = "user1";
+        ScConfig.Forward forward = new ScConfig.Forward();
+        forward.localPort = 10022;
+        forward.remoteHost = "wsl";
+        forward.remotePort = 22;
+        cfg.forwards = new ArrayList<>();
+        cfg.forwards.add(forward);
+        new ClientDemo(cfg).sync();
+    }
+
+}

_doc/img/4.jpg

@@ -0,0 +1,8 @@
+package org.wowtools.hppt.kafkademo;
+
+/**
+ * @author liuyu
+ * @date 2024/6/15
+ */
+public class KafkaCtx {
+}

_doc/img/kafkademo.jpg

@@ -0,0 +1,82 @@
+package org.wowtools.hppt.kafkademo;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+
+/**
+ * kafka工具类
+ *
+ * @author liuyu
+ * @date 2024/6/15
+ */
+public class KafkaUtil {
+
+    //客户端发数据的topic
+    public static final String ClientSendTopic = "client-send";
+    //服务端发数据的topic
+    public static final String ServerSendTopic = "server-send";
+
+
+    //基本的kafka连接配置
+    private static Properties buildProperties() {
+        Properties props = new Properties();
+        props.put("bootstrap.servers", "wsl:9092"); // 部署在电脑C上的Kafka服务器地址
+        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
+        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+        return props;
+    }
+
+    @FunctionalInterface
+    public interface BytesFunction {
+        void f(byte[] bytes);
+    }
+
+    /**
+     * 构造一个向指定topic发送bytes数据的工具
+     *
+     * @param topic 主题
+     * @return BytesFunction 调用其f(byte[] bytes)方法发送数据
+     */
+    public static BytesFunction buildProducer(String topic) {
+        Producer<String, byte[]> producer = new KafkaProducer<>(buildProperties());
+        return (bytes -> {
+            ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, bytes);
+            producer.send(record);
+        });
+    }
+
+    /**
+     * 消费kafka数据
+     *
+     * @param groupId 消费者组
+     * @param topic   主题
+     * @param cb      消费到字节时回调
+     */
+    public static void buildConsumer(String groupId, String topic, BytesFunction cb) {
+        Properties props = buildProperties();
+        props.put("group.id", groupId+System.currentTimeMillis());
+        KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
+        // 订阅主题
+        consumer.subscribe(Collections.singletonList(topic));
+        // 消费消息
+        Thread.startVirtualThread(()->{
+            while (true) {
+                ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(10));
+                for (ConsumerRecord<String, byte[]> record : records) {
+                    byte[] value = record.value();
+                    cb.f(value);
+                }
+            }
+        });
+    }
+}