邓耀骏 / Software_Architecture_Experiment_onetwofour · 提交

转到一个项目

作者 邓耀骏
提交 a2b71c3f5eb63d52c0385ea3ea04c4817b87159c 1 个父辈 defc17d6

feat(two_c):实现二更改为C实现

内嵌 并排对比
正在显示 14 个修改的文件 包含 809 行增加 和 4 行删除
  1 +<component name="ArtifactManager">
  2 + <artifact type="jar" name="client:jar">
  3 + <output-path>$PROJECT_DIR$/out/artifacts/client_jar</output-path>
  4 + <root id="archive" name="Software_Architecture_Experiment_onetwoFour.jar">
  5 + <element id="module-output" name="Software_Architecture_Experiment_onetwoFour" />
  6 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-dns/4.1.92.Final/netty-codec-dns-4.1.92.Final.jar" path-in-jar="/" />
  7 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-classes-kqueue/4.1.92.Final/netty-transport-classes-kqueue-4.1.92.Final.jar" path-in-jar="/" />
  8 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-sctp/4.1.92.Final/netty-transport-sctp-4.1.92.Final.jar" path-in-jar="/" />
  9 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver-dns-classes-macos/4.1.92.Final/netty-resolver-dns-classes-macos-4.1.92.Final.jar" path-in-jar="/" />
  10 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver-dns-native-macos/4.1.92.Final/netty-resolver-dns-native-macos-4.1.92.Final-osx-aarch_64.jar" path-in-jar="/" />
  11 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-socks/4.1.92.Final/netty-codec-socks-4.1.92.Final.jar" path-in-jar="/" />
  12 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/slf4j/slf4j-api/1.7.36/slf4j-api-1.7.36.jar" path-in-jar="/" />
  13 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/eclipse/paho/org.eclipse.paho.client.mqttv3/1.2.5/org.eclipse.paho.client.mqttv3-1.2.5.jar" path-in-jar="/" />
  14 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-http2/4.1.92.Final/netty-codec-http2-4.1.92.Final.jar" path-in-jar="/" />
  15 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-epoll/4.1.92.Final/netty-transport-native-epoll-4.1.92.Final-linux-x86_64.jar" path-in-jar="/" />
  16 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/com/github/luben/zstd-jni/1.5.2-1/zstd-jni-1.5.2-1.jar" path-in-jar="/" />
  17 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-handler-ssl-ocsp/4.1.92.Final/netty-handler-ssl-ocsp-4.1.92.Final.jar" path-in-jar="/" />
  18 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport/4.1.92.Final/netty-transport-4.1.92.Final.jar" path-in-jar="/" />
  19 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-unix-common/4.1.92.Final/netty-transport-native-unix-common-4.1.92.Final.jar" path-in-jar="/" />
  20 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-rxtx/4.1.92.Final/netty-transport-rxtx-4.1.92.Final.jar" path-in-jar="/" />
  21 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-udt/4.1.92.Final/netty-transport-udt-4.1.92.Final.jar" path-in-jar="/" />
  22 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-kqueue/4.1.92.Final/netty-transport-native-kqueue-4.1.92.Final-osx-aarch_64.jar" path-in-jar="/" />
  23 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver/4.1.92.Final/netty-resolver-4.1.92.Final.jar" path-in-jar="/" />
  24 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-stomp/4.1.92.Final/netty-codec-stomp-4.1.92.Final.jar" path-in-jar="/" />
  25 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-common/4.1.92.Final/netty-common-4.1.92.Final.jar" path-in-jar="/" />
  26 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-memcache/4.1.92.Final/netty-codec-memcache-4.1.92.Final.jar" path-in-jar="/" />
  27 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/xerial/snappy/snappy-java/1.1.8.4/snappy-java-1.1.8.4.jar" path-in-jar="/" />
  28 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver-dns-native-macos/4.1.92.Final/netty-resolver-dns-native-macos-4.1.92.Final-osx-x86_64.jar" path-in-jar="/" />
  29 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-buffer/4.1.92.Final/netty-buffer-4.1.92.Final.jar" path-in-jar="/" />
  30 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-mqtt/4.1.92.Final/netty-codec-mqtt-4.1.92.Final.jar" path-in-jar="/" />
  31 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-redis/4.1.92.Final/netty-codec-redis-4.1.92.Final.jar" path-in-jar="/" />
  32 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-handler/4.1.92.Final/netty-handler-4.1.92.Final.jar" path-in-jar="/" />
  33 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-kqueue/4.1.92.Final/netty-transport-native-kqueue-4.1.92.Final-osx-x86_64.jar" path-in-jar="/" />
  34 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec/4.1.92.Final/netty-codec-4.1.92.Final.jar" path-in-jar="/" />
  35 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-all/4.1.92.Final/netty-all-4.1.92.Final.jar" path-in-jar="/" />
  36 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/lz4/lz4-java/1.8.0/lz4-java-1.8.0.jar" path-in-jar="/" />
  37 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/apache/kafka/kafka-clients/3.2.3/kafka-clients-3.2.3.jar" path-in-jar="/" />
  38 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver-dns/4.1.92.Final/netty-resolver-dns-4.1.92.Final.jar" path-in-jar="/" />
  39 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-haproxy/4.1.92.Final/netty-codec-haproxy-4.1.92.Final.jar" path-in-jar="/" />
  40 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-epoll/4.1.92.Final/netty-transport-native-epoll-4.1.92.Final-linux-aarch_64.jar" path-in-jar="/" />
  41 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-xml/4.1.92.Final/netty-codec-xml-4.1.92.Final.jar" path-in-jar="/" />
  42 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-handler-proxy/4.1.92.Final/netty-handler-proxy-4.1.92.Final.jar" path-in-jar="/" />
  43 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-smtp/4.1.92.Final/netty-codec-smtp-4.1.92.Final.jar" path-in-jar="/" />
  44 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-classes-epoll/4.1.92.Final/netty-transport-classes-epoll-4.1.92.Final.jar" path-in-jar="/" />
  45 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-http/4.1.92.Final/netty-codec-http-4.1.92.Final.jar" path-in-jar="/" />
  46 + </root>
  47 + </artifact>
  48 +</component>
  1 +<component name="ArtifactManager">
  2 + <artifact type="jar" name="server:jar">
  3 + <output-path>$PROJECT_DIR$/out/artifacts/server_jar</output-path>
  4 + <root id="archive" name="Software_Architecture_Experiment_onetwoFour.jar">
  5 + <element id="module-output" name="Software_Architecture_Experiment_onetwoFour" />
  6 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-dns/4.1.92.Final/netty-codec-dns-4.1.92.Final.jar" path-in-jar="/" />
  7 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-classes-kqueue/4.1.92.Final/netty-transport-classes-kqueue-4.1.92.Final.jar" path-in-jar="/" />
  8 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-sctp/4.1.92.Final/netty-transport-sctp-4.1.92.Final.jar" path-in-jar="/" />
  9 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver-dns-classes-macos/4.1.92.Final/netty-resolver-dns-classes-macos-4.1.92.Final.jar" path-in-jar="/" />
  10 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver-dns-native-macos/4.1.92.Final/netty-resolver-dns-native-macos-4.1.92.Final-osx-aarch_64.jar" path-in-jar="/" />
  11 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-socks/4.1.92.Final/netty-codec-socks-4.1.92.Final.jar" path-in-jar="/" />
  12 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/slf4j/slf4j-api/1.7.36/slf4j-api-1.7.36.jar" path-in-jar="/" />
  13 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/eclipse/paho/org.eclipse.paho.client.mqttv3/1.2.5/org.eclipse.paho.client.mqttv3-1.2.5.jar" path-in-jar="/" />
  14 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-http2/4.1.92.Final/netty-codec-http2-4.1.92.Final.jar" path-in-jar="/" />
  15 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-epoll/4.1.92.Final/netty-transport-native-epoll-4.1.92.Final-linux-x86_64.jar" path-in-jar="/" />
  16 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/com/github/luben/zstd-jni/1.5.2-1/zstd-jni-1.5.2-1.jar" path-in-jar="/" />
  17 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-handler-ssl-ocsp/4.1.92.Final/netty-handler-ssl-ocsp-4.1.92.Final.jar" path-in-jar="/" />
  18 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport/4.1.92.Final/netty-transport-4.1.92.Final.jar" path-in-jar="/" />
  19 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-unix-common/4.1.92.Final/netty-transport-native-unix-common-4.1.92.Final.jar" path-in-jar="/" />
  20 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-rxtx/4.1.92.Final/netty-transport-rxtx-4.1.92.Final.jar" path-in-jar="/" />
  21 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-udt/4.1.92.Final/netty-transport-udt-4.1.92.Final.jar" path-in-jar="/" />
  22 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-kqueue/4.1.92.Final/netty-transport-native-kqueue-4.1.92.Final-osx-aarch_64.jar" path-in-jar="/" />
  23 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver/4.1.92.Final/netty-resolver-4.1.92.Final.jar" path-in-jar="/" />
  24 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-stomp/4.1.92.Final/netty-codec-stomp-4.1.92.Final.jar" path-in-jar="/" />
  25 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-common/4.1.92.Final/netty-common-4.1.92.Final.jar" path-in-jar="/" />
  26 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-memcache/4.1.92.Final/netty-codec-memcache-4.1.92.Final.jar" path-in-jar="/" />
  27 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/xerial/snappy/snappy-java/1.1.8.4/snappy-java-1.1.8.4.jar" path-in-jar="/" />
  28 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver-dns-native-macos/4.1.92.Final/netty-resolver-dns-native-macos-4.1.92.Final-osx-x86_64.jar" path-in-jar="/" />
  29 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-buffer/4.1.92.Final/netty-buffer-4.1.92.Final.jar" path-in-jar="/" />
  30 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-mqtt/4.1.92.Final/netty-codec-mqtt-4.1.92.Final.jar" path-in-jar="/" />
  31 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-redis/4.1.92.Final/netty-codec-redis-4.1.92.Final.jar" path-in-jar="/" />
  32 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-handler/4.1.92.Final/netty-handler-4.1.92.Final.jar" path-in-jar="/" />
  33 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-kqueue/4.1.92.Final/netty-transport-native-kqueue-4.1.92.Final-osx-x86_64.jar" path-in-jar="/" />
  34 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec/4.1.92.Final/netty-codec-4.1.92.Final.jar" path-in-jar="/" />
  35 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-all/4.1.92.Final/netty-all-4.1.92.Final.jar" path-in-jar="/" />
  36 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/lz4/lz4-java/1.8.0/lz4-java-1.8.0.jar" path-in-jar="/" />
  37 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/org/apache/kafka/kafka-clients/3.2.3/kafka-clients-3.2.3.jar" path-in-jar="/" />
  38 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-resolver-dns/4.1.92.Final/netty-resolver-dns-4.1.92.Final.jar" path-in-jar="/" />
  39 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-haproxy/4.1.92.Final/netty-codec-haproxy-4.1.92.Final.jar" path-in-jar="/" />
  40 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-native-epoll/4.1.92.Final/netty-transport-native-epoll-4.1.92.Final-linux-aarch_64.jar" path-in-jar="/" />
  41 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-xml/4.1.92.Final/netty-codec-xml-4.1.92.Final.jar" path-in-jar="/" />
  42 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-handler-proxy/4.1.92.Final/netty-handler-proxy-4.1.92.Final.jar" path-in-jar="/" />
  43 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-smtp/4.1.92.Final/netty-codec-smtp-4.1.92.Final.jar" path-in-jar="/" />
  44 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-transport-classes-epoll/4.1.92.Final/netty-transport-classes-epoll-4.1.92.Final.jar" path-in-jar="/" />
  45 + <element id="extracted-dir" path="$MAVEN_REPOSITORY$/io/netty/netty-codec-http/4.1.92.Final/netty-codec-http-4.1.92.Final.jar" path-in-jar="/" />
  46 + </root>
  47 + </artifact>
  48 +</component>
@@ -14,6 +14,18 @@ @@ -14,6 +14,18 @@
14 <version>4.1.92.Final</version> 14 <version>4.1.92.Final</version>
15 <scope>compile</scope> 15 <scope>compile</scope>
16 </dependency> 16 </dependency>
  17 +
  18 + <dependency>
  19 + <groupId>org.apache.kafka</groupId>
  20 + <artifactId>kafka-clients</artifactId>
  21 + <version>3.2.3</version>
  22 + </dependency>
  23 +
  24 + <dependency>
  25 + <groupId>org.eclipse.paho</groupId>
  26 + <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  27 + <version>1.2.5</version>
  28 + </dependency>
17 </dependencies> 29 </dependencies>
18 30
19 31
  1 +package four;
  2 +
  3 +import org.eclipse.paho.client.mqttv3.MqttException;
  4 +
  5 +import java.util.ArrayList;
  6 +import java.util.List;
  7 +import java.util.concurrent.Executors;
  8 +import java.util.concurrent.TimeUnit;
  9 +
  10 +public class Main {
  11 + public static void main(String[] args) throws MqttException {
  12 + String brokerUrl = "tcp://localhost:9092";
  13 + String topic = "topic";
  14 +
  15 + List<MqttProducerClient> producers = new ArrayList<>();
  16 + for (int i = 0; i < 3; i++) {
  17 + String clientId = "producer-" + i;
  18 + MqttProducerClient client = new MqttProducerClient(clientId, brokerUrl, topic);
  19 + producers.add(client);
  20 + }
  21 +
  22 + List<MqttConsumerClient> consumers = new ArrayList<>();
  23 + for (int i = 0; i < 3; i++) {
  24 + String clientId = "consumer-" + i;
  25 + MqttConsumerClient client = new MqttConsumerClient(clientId, brokerUrl, topic);
  26 + consumers.add(client);
  27 + }
  28 +
  29 + Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
  30 + for (MqttProducerClient client : producers) {
  31 + try {
  32 + String payload = "{\"temperature\":" + Math.random() * 100 + ",\"humidity\":" + Math.random() * 100 + ",\"wind_speed\":" + Math.random() * 100 + "}";
  33 + client.publish(payload);
  34 + } catch (MqttException e) {
  35 + e.printStackTrace();
  36 + }
  37 + }
  38 + }, 0, 1, TimeUnit.SECONDS);
  39 +
  40 + Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
  41 + for (MqttConsumerClient client : consumers) {
  42 + try {
  43 + client.consume();
  44 + } catch (MqttException e) {
  45 + e.printStackTrace();
  46 + }
  47 + }
  48 + }, 0, 1, TimeUnit.SECONDS);
  49 + }
  50 +}
  1 +package four;
  2 +
  3 +
  4 +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
  5 +import org.eclipse.paho.client.mqttv3.MqttCallback;
  6 +import org.eclipse.paho.client.mqttv3.MqttMessage;
  7 +
  8 +public class MqttConsumerCallback implements MqttCallback {
  9 + @Override
  10 + public void messageArrived(String topic, MqttMessage message) throws Exception {
  11 + System.out.println("Received message: " + message.toString());
  12 + }
  13 +
  14 + @Override
  15 + public void deliveryComplete(IMqttDeliveryToken token) {
  16 + // do nothing
  17 + }
  18 +
  19 + @Override
  20 + public void connectionLost(Throwable cause) {
  21 + // do nothing
  22 + }
  23 +}
  1 +package four;
  2 +
  3 +
  4 +import io.netty.channel.EventLoopGroup;
  5 +import io.netty.channel.nio.NioEventLoopGroup;
  6 +import org.apache.kafka.clients.consumer.ConsumerRecord;
  7 +import org.apache.kafka.clients.consumer.ConsumerRecords;
  8 +import org.apache.kafka.clients.consumer.KafkaConsumer;
  9 +import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
  10 +import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  11 +import org.eclipse.paho.client.mqttv3.MqttException;
  12 +import org.eclipse.paho.client.mqttv3.MqttMessage;
  13 +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
  14 +
  15 +import java.time.Duration;
  16 +import java.util.Collections;
  17 +import java.util.Properties;
  18 +
  19 +public class MqttConsumerClient {
  20 + // Kafka消费者客户端
  21 + private final KafkaConsumer<String, String> kafkaConsumer;
  22 + // Netty MQTT客户端
  23 + private final MqttAsyncClient mqttClient;
  24 +
  25 + public MqttConsumerClient(String clientId, String brokerUrl, String topic) throws MqttException {
  26 + // 初始化Kafka消费者客户端
  27 + Properties props = new Properties();
  28 + props.put("bootstrap.servers", brokerUrl);
  29 + props.put("group.id", clientId);
  30 + props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  31 + props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  32 + this.kafkaConsumer = new KafkaConsumer<>(props);
  33 + this.kafkaConsumer.subscribe(Collections.singletonList(topic));
  34 +
  35 + // 初始化Netty MQTT客户端
  36 + EventLoopGroup group = new NioEventLoopGroup();
  37 + this.mqttClient = new MqttAsyncClient(brokerUrl, clientId, new MemoryPersistence());
  38 + this.mqttClient.setCallback(new MqttConsumerCallback());
  39 + this.mqttClient.connect(new MqttConnectOptions(), null).waitForCompletion();
  40 + this.mqttClient.subscribe(topic, 1).waitForCompletion();
  41 + }
  42 +
  43 + public void consume() throws MqttException {
  44 + ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1));
  45 + for (ConsumerRecord<String, String> record : records) {
  46 + String payload = record.value();
  47 + MqttMessage message = new MqttMessage(payload.getBytes());
  48 + message.setQos(1);
  49 + mqttClient.publish("topic", message).waitForCompletion();
  50 + }
  51 + }
  52 +
  53 + public void close() throws MqttException {
  54 + kafkaConsumer.close();
  55 + mqttClient.disconnect().waitForCompletion();
  56 + mqttClient.close();
  57 + }
  58 +}
  1 +package four;
  2 +
  3 +
  4 +import org.apache.kafka.clients.producer.KafkaProducer;
  5 +import org.apache.kafka.clients.producer.ProducerRecord;
  6 +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
  7 +import org.eclipse.paho.client.mqttv3.MqttCallback;
  8 +import org.eclipse.paho.client.mqttv3.MqttMessage;
  9 +
  10 +public class MqttProducerCallback implements MqttCallback {
  11 + private final KafkaProducer<String, String> kafkaProducer;
  12 + private final String topic;
  13 +
  14 + public MqttProducerCallback(KafkaProducer<String, String> kafkaProducer, String topic) {
  15 + this.kafkaProducer = kafkaProducer;
  16 + this.topic = topic;
  17 + }
  18 +
  19 + @Override
  20 + public void messageArrived(String topic, MqttMessage message) throws Exception {
  21 + kafkaProducer.send(new ProducerRecord<>(this.topic, message.toString()));
  22 + }
  23 +
  24 + @Override
  25 + public void deliveryComplete(IMqttDeliveryToken token) {
  26 + // do nothing
  27 + }
  28 +
  29 + @Override
  30 + public void connectionLost(Throwable cause) {
  31 + // do nothing
  32 + }
  33 +}
  1 +package four;
  2 +
  3 +import io.netty.channel.EventLoopGroup;
  4 +import io.netty.channel.nio.NioEventLoopGroup;
  5 +import org.apache.kafka.clients.producer.KafkaProducer;
  6 +import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
  7 +import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  8 +import org.eclipse.paho.client.mqttv3.MqttException;
  9 +import org.eclipse.paho.client.mqttv3.MqttMessage;
  10 +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
  11 +
  12 +import java.util.Properties;
  13 +
  14 +public class MqttProducerClient {
  15 + // Kafka生产者客户端
  16 + private final KafkaProducer<String, String> kafkaProducer;
  17 + // Netty MQTT客户端
  18 + private final MqttAsyncClient mqttClient;
  19 +
  20 + public MqttProducerClient(String clientId, String brokerUrl, String topic) throws MqttException {
  21 + // 初始化Kafka生产者客户端
  22 + Properties props = new Properties();
  23 + props.put("bootstrap.servers", brokerUrl);
  24 + props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  25 + props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  26 + this.kafkaProducer = new KafkaProducer<>(props);
  27 +
  28 + // 初始化Netty MQTT客户端
  29 + EventLoopGroup group = new NioEventLoopGroup();
  30 + this.mqttClient = new MqttAsyncClient(brokerUrl, clientId, new MemoryPersistence());
  31 + this.mqttClient.setCallback(new MqttProducerCallback(kafkaProducer, topic));
  32 + this.mqttClient.connect(new MqttConnectOptions(), null).waitForCompletion();
  33 + this.mqttClient.subscribe(topic, 1).waitForCompletion();
  34 + }
  35 +
  36 + public void publish(String payload) throws MqttException {
  37 + MqttMessage message = new MqttMessage(payload.getBytes());
  38 + message.setQos(1);
  39 + mqttClient.publish("topic", message).waitForCompletion();
  40 + }
  41 +
  42 + public void close() throws MqttException {
  43 + kafkaProducer.close();
  44 + mqttClient.disconnect().waitForCompletion();
  45 + mqttClient.close();
  46 + }
  47 +}
  1 +## Netty
  2 +
  3 +Netty是一个基于Java的异步事件驱动的网络应用程序框架。它提供了一种高效、可扩展和可靠的网络编程解决方案，可以帮助开发者快速构建高性能、可靠的网络应用程序。
  4 +
  5 +Netty的优点包括：
  6 +
  7 +1. 高性能：Netty采用NIO（非阻塞I/O）模型，可以实现高并发、高吞吐量的网络通信，同时减少了线程的创建和销毁，提高了系统的性能。
  8 +
  9 +2. 可扩展性：Netty提供了一套灵活的事件处理机制，可以方便地扩展和定制网络应用程序，同时支持多种协议和编解码器。
  10 +
  11 +3. 可靠性：Netty支持TCP协议的可靠传输，可以自动重传丢失的数据包，保证了数据的可靠性和完整性。
  12 +
  13 +4. 易用性：Netty提供了简洁且易于使用的API，同时支持多种编程模型，包括阻塞、非阻塞和事件驱动等，可以满足不同场景下的需求。
  14 +
  15 +5. 安全性：Netty提供了一套完整的SSL/TLS支持，可以保证网络通信的安全性。
  16 +
  17 +总之，Netty是一个功能强大、易于使用、可靠性高的网络编程框架，可以帮助开发者快速构建高性能、可靠的网络应用程序。
  18 +
  19 +## Kafka
  20 +Kafka是一个分布式的、可扩展的、高吞吐量的消息队列系统。它最初由LinkedIn公司开发，现在已经成为Apache软件基金会的顶级项目之一。
  21 +
  22 +Kafka的核心设计思想是将消息的生产者和消费者解耦，通过消息队列来实现异步通信。Kafka的消息以topic为单位进行分类和存储，生产者将消息发送到特定的topic，消费者从topic中读取消息。Kafka支持多个消费者同时消费同一个topic，同时支持消息的持久化存储和高可靠性的数据复制机制。
  23 +
  24 +Kafka的优点包括：
  25 +
  26 +1. 高吞吐量：Kafka采用基于磁盘的存储方式，可以支持海量数据的存储和高吞吐量的数据读写。
  27 +
  28 +2. 可扩展性：Kafka的分布式架构可以方便地扩展集群规模，同时支持水平扩展和垂直扩展。
  29 +
  30 +3. 可靠性：Kafka采用多副本机制，可以保证消息的可靠性和数据的一致性。
  31 +
  32 +4. 灵活性：Kafka提供了灵活的消息处理机制，支持多种消息格式和编解码器。
  33 +
  34 +5. 易用性：Kafka提供了简单易用的API和管理工具，可以方便地进行配置和管理。
  35 +
  36 +总之，Kafka是一个可靠、高吞吐量的消息队列系统，可以帮助开发者构建分布式、可扩展的应用程序。Kafka广泛应用于大数据处理、实时数据流处理、日志收集和数据同步等场景。
  37 +
  38 +## MQTT
  39 +MQTT（Message Queue Telemetry Transport）协议是一种轻量级的、基于发布/订阅模式的通信协议，通常用于物联网（IoT）和嵌入式系统中的设备间通信。MQTT协议是IBM公司开发的，现在已经成为OASIS标准。
  40 +
  41 +MQTT协议的核心设计思想是将数据发布到一个主题（Topic）上，然后订阅该主题的设备可以接收到这些数据。MQTT协议支持多种消息传输质量，包括最多一次、至少一次和恰好一次，可以根据应用场景选择不同的传输质量。
  42 +
  43 +MQTT协议的优点包括：
  44 +
  45 +1. 轻量级：MQTT协议的数据包很小，传输效率高，适合在带宽有限的网络中使用。
  46 +
  47 +2. 灵活性：MQTT协议支持多种消息传输质量，可以根据应用场景选择不同的传输质量。
  48 +
  49 +3. 可靠性：MQTT协议支持消息的持久化存储和重发机制，可以保证消息的可靠性。
  50 +
  51 +4. 易用性：MQTT协议提供了简单易用的API和客户端库，可以方便地进行开发和集成。
  52 +
  53 +5. 支持广泛：MQTT协议已经成为物联网和嵌入式系统中的通信标准，得到了广泛的应用和支持。
  54 +
  55 +总之，MQTT协议是一种轻量级、灵活、可靠的通信协议，适用于物联网和嵌入式系统中的设备间通信。MQTT协议已经成为物联网领域的重要标准之一，得到了广泛的应用和支持。
  1 +## 服务端使用
  2 +
  3 +![image-20230507110716217](https://pic.lxtlovely.top/blog/202305071119289.png)
  4 +
  5 +1. `cd /home/dyj`
  6 +2. `gcc server.c -o server` 编译为可执行文件 server
  7 +3. `gcc client.c -o client` 编译为可执行文件 client
  8 +4. `./server 8887` 开启服务端并设置端口为 ：8877
  9 +
  10 +
  11 +
  12 +## 客户端
  13 +
  14 +![image-20230507111913598](https://pic.lxtlovely.top/blog/202305071143981.png)
  15 +
  16 +
  17 +
  18 +1. `cd /home/dyj`
  19 +2. `./client 111.230.239.199 8887 f1` ./app <IP地址> <端口号> <名称>
  20 +
  21 +
  22 +
  23 +![image-20230507111935188](https://pic.lxtlovely.top/blog/202305071143800.png)
  24 +
  25 +1. `cd /home/dyj`
  26 +2. `./client 111.230.239.199 8887 f2`
  27 +
  1 +#include <stdio.h>
  2 +#include <unistd.h>
  3 +#include <string.h>
  4 +#include <sys/types.h>
  5 +#include <sys/stat.h>
  6 +#include <fcntl.h>
  7 +#include <dirent.h>
  8 +#include <stdlib.h>
  9 +#include <pthread.h>
  10 +#include <semaphore.h>
  11 +#include <signal.h>
  12 +#include <sys/socket.h>
  13 +#include <arpa/inet.h>
  14 +#include <netinet/in.h>
  15 +#include <pthread.h>
  16 +#include <sys/select.h>
  17 +#include <sys/time.h>
  18 +#include <poll.h>
  19 +#include <sys/epoll.h>
  20 +
  21 +//消息结构体
  22 +struct MSG_DATA
  23 +{
  24 + char type; //消息类型. 0表示有聊天的消息数据 1表示好友上线 2表示好友下线
  25 + char name[50]; //好友名称
  26 + int number; //在线人数的数量
  27 + unsigned char buff[100]; //发送的聊天数据消息
  28 +};
  29 +struct MSG_DATA msg_data;
  30 +
  31 +#define MAX_EVENTS 2
  32 +struct epoll_event ev, events[MAX_EVENTS];
  33 +int epollfd;
  34 +int nfds;
  35 +
  36 +//文件接收端
  37 +int main(int argc,char **argv)
  38 +{
  39 + if(argc!=4)
  40 + {
  41 + printf("./app <IP地址> <端口号> <名称>\n");
  42 + return 0;
  43 + }
  44 + int sockfd;
  45 + //忽略 SIGPIPE 信号--方式服务器向无效的套接字写数据导致进程退出
  46 + signal(SIGPIPE,SIG_IGN);
  47 +
  48 + /*1. 创建socket套接字*/
  49 + sockfd=socket(AF_INET,SOCK_STREAM,0);
  50 + /*2. 连接服务器*/
  51 + struct sockaddr_in addr;
  52 + addr.sin_family=AF_INET;
  53 + addr.sin_port=htons(atoi(argv[2])); // 端口号0~65535
  54 + addr.sin_addr.s_addr=inet_addr(argv[1]); //IP地址
  55 + if(connect(sockfd,(const struct sockaddr *)&addr,sizeof(struct sockaddr_in))!=0)
  56 + {
  57 + printf("客户端:服务器连接失败.\n");
  58 + return 0;
  59 + }
  60 +
  61 + /*3. 发送消息表示上线*/
  62 + msg_data.type=1;
  63 + strcpy(msg_data.name,argv[3]);
  64 + write(sockfd,&msg_data,sizeof(struct MSG_DATA));
  65 +
  66 + int cnt;
  67 + int i;
  68 + //创建专用文件描述符
  69 + epollfd = epoll_create(10);
  70 + //添加要监听的文件描述符
  71 + ev.events = EPOLLIN;
  72 + ev.data.fd = sockfd;
  73 + epoll_ctl(epollfd, EPOLL_CTL_ADD, sockfd, &ev);
  74 +
  75 + ev.events = EPOLLIN;
  76 + ev.data.fd = 0; //标准输入文件描述符
  77 + epoll_ctl(epollfd, EPOLL_CTL_ADD, 0, &ev);
  78 +
  79 + while(1)
  80 + {
  81 + //监听事件
  82 + nfds=epoll_wait(epollfd,events,MAX_EVENTS,-1);
  83 + if(nfds)
  84 + {
  85 + for(i=0;i<nfds;i++)
  86 + {
  87 + if(events[i].data.fd==sockfd) //判断收到服务器的消息
  88 + {
  89 + cnt=read(sockfd,&msg_data,sizeof(struct MSG_DATA));
  90 + if(cnt<=0) //判断服务器是否断开了连接
  91 + {
  92 + printf("服务器已经退出.\n");
  93 + goto SERVER_ERROR;
  94 + }
  95 + else if(cnt>0)
  96 + {
  97 + if(msg_data.type==0)
  98 + {
  99 + printf("%s:%s 在线人数:%d\n",msg_data.name,msg_data.buff,msg_data.number);
  100 + }
  101 + else if(msg_data.type==1)
  102 + {
  103 + printf("%s 好友上线. 在线人数:%d\n",msg_data.name,msg_data.number);
  104 + }
  105 + else if(msg_data.type==2)
  106 + {
  107 + printf("%s 好友下线. 在线人数:%d\n",msg_data.name,msg_data.number);
  108 + }
  109 + }
  110 + }
  111 + else if(events[i].data.fd==0) //表示键盘上有数据输入
  112 + {
  113 + gets(msg_data.buff); //读取键盘上的消息
  114 + msg_data.type=0; //表示正常消息
  115 + strcpy(msg_data.name,argv[3]); //名称
  116 + write(sockfd,&msg_data,sizeof(struct MSG_DATA));
  117 + }
  118 + }
  119 + }
  120 + }
  121 +SERVER_ERROR:
  122 + close(sockfd);
  123 + return 0;
  124 +}
  1 +#include <stdio.h>
  2 +#include <unistd.h>
  3 +#include <string.h>
  4 +#include <sys/types.h>
  5 +#include <sys/stat.h>
  6 +#include <fcntl.h>
  7 +#include <dirent.h>
  8 +#include <stdlib.h>
  9 +#include <pthread.h>
  10 +#include <semaphore.h>
  11 +#include <signal.h>
  12 +#include <sys/types.h>
  13 +#include <sys/socket.h>
  14 +#include <arpa/inet.h>
  15 +#include <netinet/in.h>
  16 +#include <pthread.h>
  17 +#include <sys/select.h>
  18 +#include <sys/time.h>
  19 +#include <sys/epoll.h>
  20 +
  21 +int sockfd;
  22 +//消息结构体
  23 +struct MSG_DATA
  24 +{
  25 + char type; //消息类型. 0表示有聊天的消息数据 1表示好友上线 2表示好友下线
  26 + char name[50]; //好友名称
  27 + int number; //在线人数的数量
  28 + unsigned char buff[100]; //发送的聊天数据消息
  29 +};
  30 +
  31 +//存放当前服务器连接的客户端套接字
  32 +struct CLIENT_FD
  33 +{
  34 + int fd;
  35 + char name[50]; //名称
  36 + struct CLIENT_FD *next;
  37 +};
  38 +
  39 +//定义链表头
  40 +struct CLIENT_FD *list_head=NULL;
  41 +struct CLIENT_FD *List_CreateHead(struct CLIENT_FD *list_head);
  42 +void List_AddNode(struct CLIENT_FD *list_head,int fd);
  43 +void List_DelNode(struct CLIENT_FD *list_head,int fd);
  44 +int List_GetNodeCnt(struct CLIENT_FD *list_head);
  45 +void Server_SendMsgData(struct CLIENT_FD *list_head,struct MSG_DATA *msg_data,int client_fd);
  46 +void List_SaveName(struct CLIENT_FD *list_head,struct MSG_DATA *msg_data,int client_fd);
  47 +void List_GetName(struct CLIENT_FD *list_head,struct MSG_DATA *msg_data,int client_fd);
  48 +
  49 +#define MAX_EPOLL_FD 100
  50 +struct epoll_event events[MAX_EPOLL_FD];
  51 +struct epoll_event event;
  52 +int epfd;
  53 +int nfd;
  54 +struct MSG_DATA msg_data;
  55 +
  56 +/*信号工作函数*/
  57 +void signal_work_func(int sig)
  58 +{
  59 + close(sockfd);
  60 + exit(0); //结束进程
  61 +}
  62 +
  63 +int main(int argc,char **argv)
  64 +{
  65 + if(argc!=2)
  66 + {
  67 + printf("./app <端口号>\n");
  68 + return 0;
  69 + }
  70 + signal(SIGPIPE,SIG_IGN); //忽略 SIGPIPE 信号--防止服务器异常退出
  71 + signal(SIGINT,signal_work_func);
  72 +
  73 + //创建链表头
  74 + list_head=List_CreateHead(list_head);
  75 +
  76 + /*1. 创建socket套接字*/
  77 + sockfd=socket(AF_INET,SOCK_STREAM,0);
  78 +
  79 + //设置端口号的复用功能
  80 + int on = 1;
  81 + setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
  82 +
  83 + /*2. 绑定端口号与IP地址*/
  84 + struct sockaddr_in addr;
  85 + addr.sin_family=AF_INET;
  86 + addr.sin_port=htons(atoi(argv[1])); // 端口号0~65535
  87 + addr.sin_addr.s_addr=INADDR_ANY; //inet_addr("0.0.0.0"); //IP地址
  88 + if(bind(sockfd,(const struct sockaddr *)&addr,sizeof(struct sockaddr))!=0)
  89 + {
  90 + printf("服务器:端口号绑定失败.\n");
  91 + }
  92 + /*3. 设置监听的数量*/
  93 + listen(sockfd,20);
  94 + /*4. 等待客户端连接*/
  95 + int client_fd;
  96 + struct sockaddr_in client_addr;
  97 + socklen_t addrlen;
  98 + int i;
  99 + int cnt;
  100 +
  101 + /*5. 创建epoll相关的接口*/
  102 + epfd=epoll_create(MAX_EPOLL_FD);
  103 + event.events=EPOLLIN; //监听的事件
  104 + event.data.fd=sockfd; //监听的套接字
  105 + epoll_ctl(epfd,EPOLL_CTL_ADD,sockfd,&event);
  106 + while(1)
  107 + {
  108 + //等待事件发生
  109 + nfd=epoll_wait(epfd,events,MAX_EPOLL_FD,-1);
  110 + for(i=0;i<nfd;i++)
  111 + {
  112 + if(events[i].data.fd==sockfd) //表示有新的客户端连接上服务器
  113 + {
  114 + client_fd=accept(sockfd,(struct sockaddr *)&client_addr,&addrlen);
  115 + printf("连接的客户端IP地址:%s\n",inet_ntoa(client_addr.sin_addr));
  116 + printf("连接的客户端端口号:%d\n",ntohs(client_addr.sin_port));
  117 + //保存已经连接上来的客户端
  118 + List_AddNode(list_head,client_fd);
  119 + //将新连接的客户端套接字添加到epoll函数监听队列里
  120 + event.data.fd=client_fd; //监听的套接字
  121 + epoll_ctl(epfd,EPOLL_CTL_ADD,client_fd,&event);
  122 + }
  123 + else //表示客户端给服务器发送了消息-----实现消息的转发
  124 + {
  125 + //读取客户端发送的消息
  126 + cnt=read(events[i].data.fd,&msg_data,sizeof(struct MSG_DATA));
  127 + if(cnt<=0) //表示当前客户端断开了连接
  128 + {
  129 + //获取名称
  130 + List_GetName(list_head,&msg_data,events[i].data.fd);
  131 + //删除节点
  132 + List_DelNode(list_head,events[i].data.fd);
  133 + msg_data.type=2;
  134 +
  135 + //将断开连接的客户端套接字从epoll函数监听队列里删除调用
  136 + event.data.fd=events[i].data.fd; //监听的套接字
  137 + epoll_ctl(epfd,EPOLL_CTL_DEL,events[i].data.fd,&event);
  138 + close(event.data.fd);
  139 + }
  140 + if(msg_data.type==1) //好友上线的时候保存一次名称
  141 + {
  142 + //保存名称
  143 + List_SaveName(list_head,&msg_data,events[i].data.fd);
  144 + }
  145 + //转发消息给其他好友
  146 + msg_data.number=List_GetNodeCnt(list_head); //当前在线好友人数
  147 + Server_SendMsgData(list_head,&msg_data,events[i].data.fd);
  148 + }
  149 + }
  150 + }
  151 + //退出进程
  152 + signal_work_func(0);
  153 + return 0;
  154 +}
  155 +
  156 +
  157 +/*
  158 +函数功能: 创建链表头
  159 +*/
  160 +struct CLIENT_FD *List_CreateHead(struct CLIENT_FD *list_head)
  161 +{
  162 + if(list_head==NULL)
  163 + {
  164 + list_head=malloc(sizeof(struct CLIENT_FD));
  165 + list_head->next=NULL;
  166 + }
  167 + return list_head;
  168 +}
  169 +
  170 +/*
  171 +函数功能: 添加节点
  172 +*/
  173 +void List_AddNode(struct CLIENT_FD *list_head,int fd)
  174 +{
  175 + struct CLIENT_FD *p=list_head;
  176 + struct CLIENT_FD *new_p;
  177 + while(p->next!=NULL)
  178 + {
  179 + p=p->next;
  180 + }
  181 + new_p=malloc(sizeof(struct CLIENT_FD));
  182 + new_p->next=NULL;
  183 + new_p->fd=fd;
  184 + p->next=new_p;
  185 +}
  186 +
  187 +/*
  188 +函数功能: 删除节点
  189 +*/
  190 +void List_DelNode(struct CLIENT_FD *list_head,int fd)
  191 +{
  192 + struct CLIENT_FD *p=list_head;
  193 + struct CLIENT_FD *tmp;
  194 + while(p->next!=NULL)
  195 + {
  196 + tmp=p;
  197 + p=p->next;
  198 + if(p->fd==fd) //找到了要删除的节点
  199 + {
  200 + tmp->next=p->next;
  201 + free(p);
  202 + break;
  203 + }
  204 + }
  205 +}
  206 +
  207 +/*
  208 +函数功能: 获取当前链表中有多少个节点
  209 +*/
  210 +int List_GetNodeCnt(struct CLIENT_FD *list_head)
  211 +{
  212 + int cnt=0;
  213 + struct CLIENT_FD *p=list_head;
  214 + while(p->next!=NULL)
  215 + {
  216 + p=p->next;
  217 + cnt++;
  218 + }
  219 + return cnt;
  220 +}
  221 +
  222 +/*
  223 +函数功能: 转发消息
  224 +*/
  225 +void Server_SendMsgData(struct CLIENT_FD *list_head,struct MSG_DATA *msg_data,int client_fd)
  226 +{
  227 + struct CLIENT_FD *p=list_head;
  228 + while(p->next!=NULL)
  229 + {
  230 + p=p->next;
  231 + if(p->fd!=client_fd)
  232 + {
  233 + write(p->fd,msg_data,sizeof(struct MSG_DATA));
  234 + }
  235 + }
  236 +}
  237 +
  238 +/*
  239 +函数功能: 保存好友的名称
  240 +*/
  241 +void List_SaveName(struct CLIENT_FD *list_head,struct MSG_DATA *msg_data,int client_fd)
  242 +{
  243 + struct CLIENT_FD *p=list_head;
  244 + while(p->next!=NULL)
  245 + {
  246 + p=p->next;
  247 + if(p->fd==client_fd) //找到在链表里的当前套接字
  248 + {
  249 + strcpy(p->name,msg_data->name);
  250 + }
  251 + }
  252 +}
  253 +
  254 +
  255 +/*
  256 +函数功能: 获取好友的名称
  257 +*/
  258 +void List_GetName(struct CLIENT_FD *list_head,struct MSG_DATA *msg_data,int client_fd)
  259 +{
  260 + struct CLIENT_FD *p=list_head;
  261 + while(p->next!=NULL)
  262 + {
  263 + p=p->next;
  264 + if(p->fd==client_fd) //找到在链表里的当前套接字
  265 + {
  266 + strcpy(msg_data->name,p->name);
  267 + }
  268 + }
  269 +}
@@ -9,26 +9,29 @@ import java.util.Scanner; @@ -9,26 +9,29 @@ import java.util.Scanner;
9 public class Client { 9 public class Client {
10 private static final int BUFFER_SIZE = 1024; 10 private static final int BUFFER_SIZE = 1024;
11 private static final String HOST = "localhost"; 11 private static final String HOST = "localhost";
12 - private static final int PORT = 8888; 12 + private static final int PORT = 8877;
13 13
14 public static void main(String[] args) { 14 public static void main(String[] args) {
15 try { 15 try {
  16 + // 创建SocketChannel对象，连接服务器。
16 SocketChannel clientChannel = SocketChannel.open(); 17 SocketChannel clientChannel = SocketChannel.open();
17 clientChannel.configureBlocking(false); 18 clientChannel.configureBlocking(false);
18 clientChannel.connect(new InetSocketAddress(HOST, PORT)); 19 clientChannel.connect(new InetSocketAddress(HOST, PORT));
19 20
20 while (!clientChannel.finishConnect()) { 21 while (!clientChannel.finishConnect()) {
21 - // wait until connection is established 22 + // 如果连接未完成，等待连接完成。
22 } 23 }
23 24
24 System.out.println("Connected to server " + HOST + ":" + PORT); 25 System.out.println("Connected to server " + HOST + ":" + PORT);
25 System.out.println("Enter your name:"); 26 System.out.println("Enter your name:");
26 27
  28 + // 读取用户输入的用户名，将其发送给服务器。
27 Scanner scanner = new Scanner(System.in); 29 Scanner scanner = new Scanner(System.in);
28 String name = scanner.nextLine(); 30 String name = scanner.nextLine();
29 ByteBuffer buffer = ByteBuffer.wrap(name.getBytes()); 31 ByteBuffer buffer = ByteBuffer.wrap(name.getBytes());
30 clientChannel.write(buffer); 32 clientChannel.write(buffer);
31 33
  34 + // 创建一个新线程，使用SocketChannel的read()方法读取服务器发送的消息，并在控制台上显示。
32 new Thread(() -> { 35 new Thread(() -> {
33 while (true) { 36 while (true) {
34 try { 37 try {
@@ -44,6 +47,7 @@ public class Client { @@ -44,6 +47,7 @@ public class Client {
44 } 47 }
45 }).start(); 48 }).start();
46 49
  50 + // 在主线程中，读取用户输入的消息，使用SocketChannel的write()方法将其发送给服务器。
47 while (true) { 51 while (true) {
48 String message = scanner.nextLine(); 52 String message = scanner.nextLine();
49 buffer = ByteBuffer.wrap(message.getBytes()); 53 buffer = ByteBuffer.wrap(message.getBytes());
@@ -13,7 +13,7 @@ import java.util.Map; @@ -13,7 +13,7 @@ import java.util.Map;
13 13
14 public class Server { 14 public class Server {
15 private static final int BUFFER_SIZE = 1024; 15 private static final int BUFFER_SIZE = 1024;
16 - private static final int PORT = 8888; 16 + private static final int PORT = 8877;
17 17
18 private Selector selector; 18 private Selector selector;
19 private Map<SocketChannel, String> clientMap = new HashMap<>(); 19 private Map<SocketChannel, String> clientMap = new HashMap<>();
@@ -25,6 +25,7 @@ public class Server { @@ -25,6 +25,7 @@ public class Server {
25 25
26 public void startServer() { 26 public void startServer() {
27 try { 27 try {
  28 + // 创建Selector对象和ServerSocketChannel对象，并将ServerSocketChannel注册到Selector上，监听连接事件。
28 selector = Selector.open(); 29 selector = Selector.open();
29 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); 30 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
30 serverSocketChannel.configureBlocking(false); 31 serverSocketChannel.configureBlocking(false);
@@ -32,16 +33,21 @@ public class Server { @@ -32,16 +33,21 @@ public class Server {
32 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); 33 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
33 System.out.println("Server started on port " + PORT); 34 System.out.println("Server started on port " + PORT);
34 35
  36 + // 进入主循环
35 while (true) { 37 while (true) {
  38 + // 调用Selector的select()方法等待事件发生
36 int readyChannels = selector.select(); 39 int readyChannels = selector.select();
37 if (readyChannels == 0) { 40 if (readyChannels == 0) {
38 continue; 41 continue;
39 } 42 }
40 43
  44 + // 如果有事件发生，使用迭代器遍历SelectionKey集合，处理每个事件。
41 Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator(); 45 Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
42 while (keyIterator.hasNext()) { 46 while (keyIterator.hasNext()) {
43 SelectionKey key = keyIterator.next(); 47 SelectionKey key = keyIterator.next();
44 48
  49 + // 如果是连接事件，使用ServerSocketChannel的accept()方法接受客户端连接，
  50 + // 并将客户端的SocketChannel注册到Selector上，监听读事件。
45 if (key.isAcceptable()) { 51 if (key.isAcceptable()) {
46 SocketChannel clientChannel = serverSocketChannel.accept(); 52 SocketChannel clientChannel = serverSocketChannel.accept();
47 clientChannel.configureBlocking(false); 53 clientChannel.configureBlocking(false);
@@ -50,6 +56,7 @@ public class Server { @@ -50,6 +56,7 @@ public class Server {
50 clientMap.put(clientChannel, ""); 56 clientMap.put(clientChannel, "");
51 } 57 }
52 58
  59 + // 如果是读事件，使用SocketChannel的read()方法读取客户端发送的消息，并将其转发给其他客户端。
53 if (key.isReadable()) { 60 if (key.isReadable()) {
54 SocketChannel clientChannel = (SocketChannel) key.channel(); 61 SocketChannel clientChannel = (SocketChannel) key.channel();
55 ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE); 62 ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
@@ -67,6 +74,7 @@ public class Server { @@ -67,6 +74,7 @@ public class Server {
67 } 74 }
68 } 75 }
69 76
  77 + // 处理完事件后，使用迭代器的remove()方法将SelectionKey从集合中删除。
70 keyIterator.remove(); 78 keyIterator.remove();
71 } 79 }
72 } 80 }
@@ -79,7 +87,6 @@ public class Server { @@ -79,7 +87,6 @@ public class Server {
79 87
80 for (SocketChannel clientChannel : clientMap.keySet()) { 88 for (SocketChannel clientChannel : clientMap.keySet()) {
81 if (clientChannel != senderChannel) { 89 if (clientChannel != senderChannel) {
82 - System.out.println(clientChannel);  
83 //多个对象发送，每次都得载一个新的对象，不然就只能发送一次 90 //多个对象发送，每次都得载一个新的对象，不然就只能发送一次
84 ByteBuffer buffer = ByteBuffer.wrap(message.getBytes()); 91 ByteBuffer buffer = ByteBuffer.wrap(message.getBytes());
85 clientChannel.write(buffer); 92 clientChannel.write(buffer);