diff --git a/example_output.jpg b/example_output.jpg new file mode 100644 index 0000000..b56b519 Binary files /dev/null and b/example_output.jpg differ diff --git a/file-read/src/main/java/com/wyl/file/sequence/Sequential.java b/file-read/src/main/java/com/wyl/file/sequence/Sequential.java index 955dc1e..8dbf946 100644 --- a/file-read/src/main/java/com/wyl/file/sequence/Sequential.java +++ b/file-read/src/main/java/com/wyl/file/sequence/Sequential.java @@ -7,15 +7,16 @@ import java.nio.MappedByteBuffer; import java.nio.channels.FileChannel; public class Sequential { - public static int fileWrite(String filePath, String content, int index) { + public static int fileWrite(String filePath, String content) { File file = new File(filePath); RandomAccessFile randomAccessTargetFile; MappedByteBuffer map; try { randomAccessTargetFile = new RandomAccessFile(file, "rw"); FileChannel targetFileChannel = randomAccessTargetFile.getChannel(); - map = targetFileChannel.map(FileChannel.MapMode.READ_WRITE, 0, (long) 1024 * 1024 * 1024); - map.position(index); + byte[] bytes = content.getBytes(); + map = targetFileChannel.map(FileChannel.MapMode.READ_WRITE, 0, bytes.length); + map.position(0); map.put(content.getBytes()); return map.position(); } diff --git a/file-read/src/test/java/com/wyl/file/FileTest.java b/file-read/src/test/java/com/wyl/file/FileTest.java index d5c653d..653b097 100644 --- a/file-read/src/test/java/com/wyl/file/FileTest.java +++ b/file-read/src/test/java/com/wyl/file/FileTest.java @@ -16,18 +16,17 @@ public class FileTest { public void SequentialWriteFile() { Integer index = 0; - File file = new File("./wyl.log"); MappedByteBuffer map; try (RandomAccessFile randomAccessTargetFile = new RandomAccessFile(file, "rw")) { FileChannel targetFileChannel = randomAccessTargetFile.getChannel(); - map = targetFileChannel.map(FileChannel.MapMode.READ_WRITE, pos.get(), (long) 1024 * 1024 * 1024); - if (map.isLoaded()) { - + map = targetFileChannel.map(FileChannel.MapMode.READ_WRITE, pos.get(), 2); + while (index<2) + { + map.position(index); + map.put("1".getBytes()); + index = map.position(); } - map.position(index); - map.put("123\n".getBytes()); - final int position = map.position(); } catch (Exception e) { e.printStackTrace(); diff --git a/javacv/pom.xml b/javacv/pom.xml new file mode 100644 index 0000000..c60922d --- /dev/null +++ b/javacv/pom.xml @@ -0,0 +1,26 @@ + + + + JavaBasiceDemo + com.wyl.example + 1.0-SNAPSHOT + + 4.0.0 + + javacv + + + 8 + 8 + + + + + org.bytedeco + javacv-platform + 1.5.8 + + + \ No newline at end of file diff --git a/javacv/src/main/java/com/wyl/javacv/ImageToVideoExample.java b/javacv/src/main/java/com/wyl/javacv/ImageToVideoExample.java new file mode 100644 index 0000000..af1eb61 --- /dev/null +++ b/javacv/src/main/java/com/wyl/javacv/ImageToVideoExample.java @@ -0,0 +1,52 @@ +package com.wyl.javacv; + +import org.opencv.core.Mat; +import org.opencv.core.Size; +import org.opencv.videoio.VideoWriter; + +import java.io.File; + +import static org.opencv.imgcodecs.Imgcodecs.imread; + +public class ImageToVideoExample { + + public static void main(String[] args) { + // 图像文件夹路径 + String imageFolderPath = "/Users/wyl/project/java/wyl/JavaBasiceDemo/javacv/src/main/resources/"; + + // 获取图像文件列表 + File[] imageFiles = new File(imageFolderPath).listFiles(); + + // 设置视频编码器参数 + int fourcc = VideoWriter.fourcc('X', 'V', 'I', 'D'); // 使用XVID编码器 + double fps = 24; // 视频帧率 + Mat image = imread(imageFiles[0].getAbsolutePath()); + Size frameSize = new Size(image.cols(), image.rows()); // 视频帧尺寸 + + // 创建视频写入器 + VideoWriter writer = new VideoWriter("/Users/wyl/project/java/wyl/JavaBasiceDemo/javacv/src/main/resources/output.avi", fourcc, fps, frameSize, true); + + // 逐个读取图像并写入视频 + for (File imageFile : imageFiles) { + // 加载图像 + image = imread(imageFile.getAbsolutePath()); + + // 检查图像是否加载成功 + if (image.empty()) { + System.out.println("无法加载图像：" + imageFile.getName()); + System.exit(1); + } + + // 写入视频 + writer.write(image); + + // 释放内存 + image.release(); + } + + // 释放视频写入器 + writer.release(); + + System.out.println("视频合成完成！"); + } +} diff --git a/javacv/src/main/java/com/wyl/javacv/JavaCVExample.java b/javacv/src/main/java/com/wyl/javacv/JavaCVExample.java new file mode 100644 index 0000000..59b0e34 --- /dev/null +++ b/javacv/src/main/java/com/wyl/javacv/JavaCVExample.java @@ -0,0 +1,33 @@ +package com.wyl.javacv; + + +import org.bytedeco.opencv.opencv_core.Mat; + +import static org.bytedeco.opencv.global.opencv_highgui.imshow; +import static org.bytedeco.opencv.global.opencv_highgui.waitKey; +import static org.bytedeco.opencv.global.opencv_imgcodecs.imread; +import static org.bytedeco.opencv.global.opencv_imgcodecs.imwrite; + +public class JavaCVExample { + + public static void main(String[] args) { + // 加载图像 + Mat image = imread("/Users/wyl/project/java/wyl/JavaBasiceDemo/javacv/src/main/resources/OIP.jpg"); + + // 检查图像是否加载成功 + if (image.empty()) { + System.out.println("无法加载图像"); + System.exit(1); + } + + // 显示图像 +// imshow("图像窗口", image); +// waitKey(0); + + // 保存图像 + imwrite("/Users/wyl/project/java/wyl/JavaBasiceDemo/javacv/src/main/resources/example_output.jpg", image); + + // 释放内存 + image.release(); + } +} diff --git a/javacv/src/main/resources/OIP.jpg b/javacv/src/main/resources/OIP.jpg new file mode 100644 index 0000000..a1957b4 Binary files /dev/null and b/javacv/src/main/resources/OIP.jpg differ diff --git a/javacv/src/main/resources/R.jpeg b/javacv/src/main/resources/R.jpeg new file mode 100644 index 0000000..4d785a5 Binary files /dev/null and b/javacv/src/main/resources/R.jpeg differ diff --git a/javacv/src/main/resources/example_output.jpg b/javacv/src/main/resources/example_output.jpg new file mode 100644 index 0000000..b56b519 Binary files /dev/null and b/javacv/src/main/resources/example_output.jpg differ diff --git a/kafka/src/test/java/com/wyl/kafka/test/ConsumerTest.java b/kafka/src/test/java/com/wyl/kafka/test/ConsumerTest.java index 8fc9450..6af4df2 100644 --- a/kafka/src/test/java/com/wyl/kafka/test/ConsumerTest.java +++ b/kafka/src/test/java/com/wyl/kafka/test/ConsumerTest.java @@ -6,9 +6,20 @@ import com.wyl.kafka.producers.KafkaProducerExample; import com.wyl.kafka.producers.KafkaProducerSend; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.Consumer; +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 org.junit.jupiter.api.Test; +import java.io.*; +import java.util.Arrays; +import java.util.Properties; + +import static java.nio.charset.StandardCharsets.ISO_8859_1; + @Slf4j public class ConsumerTest { final static String TOPIC_NAME = "wyl-filebeat-test"; @@ -85,4 +96,35 @@ public class ConsumerTest { KafkaConsumerPoll.PollMessageSeekEnd(consumer, TOPIC_NAME, 1); } + + @Test + public void imageSave() throws IOException { + + // 创建 Kafka 消费者配置 + Properties consumerProps = new Properties(); + consumerProps.put("bootstrap.servers", "localhost:9092"); + consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + consumerProps.put("group.id","test"); + consumerProps.put("max.request.size", "2097152"); + // 创建 Kafka 生产者和消费者 + KafkaConsumer consumer = new KafkaConsumer<>(consumerProps); + + // 接收图片消息从 Kafka + consumer.subscribe(Arrays.asList("wyl-events")); + while (true) { + ConsumerRecords records = consumer.poll(100); + for (ConsumerRecord record : records) { + byte[] imageBytes = record.value().getBytes(ISO_8859_1); + // 将图片字节数组保存到文件中 + // 这里假设图片文件名为 received-image.jpg + File receivedFile = new File("/Users/wyl/Desktop/wyl01.png"); + FileOutputStream fileOutputStream = new FileOutputStream(receivedFile); + fileOutputStream.write(imageBytes); + fileOutputStream.close(); + System.out.println(11); + } + } + } + } diff --git a/kafka/src/test/java/com/wyl/kafka/test/ProducerTest.java b/kafka/src/test/java/com/wyl/kafka/test/ProducerTest.java index 5afaf25..a51a7e7 100644 --- a/kafka/src/test/java/com/wyl/kafka/test/ProducerTest.java +++ b/kafka/src/test/java/com/wyl/kafka/test/ProducerTest.java @@ -2,10 +2,20 @@ package com.wyl.kafka.test; import com.wyl.kafka.producers.KafkaProducerExample; import com.wyl.kafka.producers.KafkaProducerSend; +import jdk.jfr.internal.tool.Main; import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; import org.junit.jupiter.api.Test; +import java.io.*; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Objects; +import java.util.Properties; + @Slf4j public class ProducerTest { final static String TOPIC_NAME = "wyl-filebeat-test"; @@ -21,4 +31,73 @@ public class ProducerTest { final Producer stringStringProducer = KafkaProducerExample.KafkaProducerTransactional(); KafkaProducerSend.sendMessageTransactional(stringStringProducer, TOPIC_NAME, "test123"); } + + @Test + public void imageSend() throws IOException, InterruptedException { + + // 创建 Kafka 生产者配置 + Properties producerProps = new Properties(); + producerProps.put("bootstrap.servers", "localhost:9092"); + producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + producerProps.put("max.request.size", "2097152"); + + // 创建 Kafka 生产者和消费者 + KafkaProducer producer = new KafkaProducer<>(producerProps); + + // 发送图片消息到 Kafka + File file = new File("/Users/wyl/Desktop/wyl.png"); + InputStream inputStream = new FileInputStream(file); + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + byte[] buffer = new byte[1024]; + int bytesRead; + while ((bytesRead = inputStream.read(buffer)) != -1) { + outputStream.write(buffer, 0, bytesRead); + } + byte[] imageData = outputStream.toByteArray(); + String s = new String(imageData,StandardCharsets.ISO_8859_1); + ProducerRecord record = new ProducerRecord<>("wyl-events", s); + producer.send(record,(m,e)->{ + if(Objects.nonNull(e)) + { + e.printStackTrace(); + } + System.out.println(m); + + }); + Thread.sleep(100000); + } + + public static void main(String[] args) throws IOException { + byte[] byteArray = { -119, 66, 67 }; + String str = new String(byteArray,StandardCharsets.ISO_8859_1); + System.out.println(str); + byte[] bytes = str.getBytes(StandardCharsets.ISO_8859_1); + System.out.println(bytes); + } + + public static void main1(String[] args) throws IOException { + File file = new File("/Users/wyl/Desktop/wyl.png"); + InputStream inputStream = new FileInputStream(file); + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + byte[] buffer = new byte[1024]; + int bytesRead; + while ((bytesRead = inputStream.read(buffer)) != -1) { + outputStream.write(buffer, 0, bytesRead); + } + byte[] imageData = outputStream.toByteArray(); + String s = new String(imageData); + } + + public static void main2(String[] args) { + byte[] byteArray = { -119, 66, 67 }; + Charset[] charsets = { StandardCharsets.UTF_8, Charset.forName("ISO-8859-1"), Charset.forName("GBK") }; + for (Charset charset : charsets) { + String str = new String(byteArray, charset); + System.out.println(charset.name() + ": " + str); + byte[] newByteArray = str.getBytes(charset); + System.out.println(charset.name() + ": " + Arrays.toString(newByteArray)); + } + + } } diff --git a/mqtt/pom.xml b/mqtt/pom.xml new file mode 100644 index 0000000..ccae63b --- /dev/null +++ b/mqtt/pom.xml @@ -0,0 +1,27 @@ + + + + JavaBasiceDemo + com.wyl.example + 1.0-SNAPSHOT + + 4.0.0 + + mqtt + + + 8 + 8 + + + + + org.eclipse.paho + org.eclipse.paho.client.mqttv3 + 1.2.5 + + + + \ No newline at end of file diff --git a/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT.java b/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT.java new file mode 100644 index 0000000..914fa29 --- /dev/null +++ b/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT.java @@ -0,0 +1,108 @@ +package com.wyl.mqtt; + +import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.TimeUnit; + +@Slf4j +public class ClientMQTT { + MqttClient client; + MqttConnectOptions options; + + public ClientMQTT() { + try { + // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存 + client = new MqttClient("tcp://127.0.0.1:61613", "wyl01", new MemoryPersistence()); + // MQTT的连接设置 + options = new MqttConnectOptions(); + // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接 + options.setCleanSession(true); + // 设置连接的用户名 + options.setUserName("admin"); + // 设置连接的密码 + options.setPassword("password".toCharArray()); + // 设置超时时间 单位为秒 + options.setConnectionTimeout(10); + // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制 + options.setKeepAliveInterval(20); + //设置自动连接 + options.setAutomaticReconnect(true); + // 设置回调 + client.setCallback(new MqttCallbackExtended() { + //连接成功回调，需要重新订阅主题 + @Override + public void connectComplete(boolean reconnect, String serverURI) { + sendNotify(true); + subscribe(); + } + + @Override + public void connectionLost(Throwable cause) { + retryConnection(3); + } + + @Override + public void deliveryComplete(IMqttDeliveryToken token) { + token.isComplete(); + } + + @Override + public void messageArrived(String topic, MqttMessage message) { + } + }); + client.connect(options); + subscribe(); + } catch (Exception e) { + log.error("mqtt异常" + e.getMessage()); + e.printStackTrace(); + } + } + + private void subscribe() { + try { + //订阅队列 + client.subscribe("wyl001", 2, (topic, message) -> { + String msg = new String(message.getPayload(), StandardCharsets.UTF_8); + log.warn("mqtt收到信息:"+msg); + //处理消息 + }); + } catch (MqttException e) { + log.error("mqtt异常" + e.getMessage()); + } + } + + private void retryConnection(int retryNumber) { + //当重试多次，依旧失败，就是机器故障，需要通知人工处理 + if (retryNumber < 0) { + sendNotify(false); + } + try { + client.reconnect(); + TimeUnit.SECONDS.sleep(30); + if (!client.isConnected()) { + retryConnection(--retryNumber); + } + } catch (MqttException | InterruptedException e) { + log.error("mqtt异常:"+e.getMessage()); + } + } + + + //发送重试通知 + private void sendNotify(boolean isSuccess) { + if (isSuccess){ + log.debug("mqtt连接成功"); + }else { + log.error("mqtt连接失败"); + } + } + + + public static void main(String[] args) { + ClientMQTT clientMQTT = new ClientMQTT(); + clientMQTT.subscribe(); + } +} \ No newline at end of file diff --git a/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT2.java b/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT2.java new file mode 100644 index 0000000..be060a2 --- /dev/null +++ b/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT2.java @@ -0,0 +1,111 @@ +package com.wyl.mqtt; + +import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.TimeUnit; + +@Slf4j +public class ClientMQTT2 { + MqttClient client; + MqttConnectOptions options; + + public ClientMQTT2() { + try { + // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存 + client = new MqttClient("tcp://127.0.0.1:61613", "wyl02", new MemoryPersistence()); + // MQTT的连接设置 + options = new MqttConnectOptions(); + // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接 + options.setCleanSession(true); + // 设置连接的用户名 + options.setUserName("admin"); + // 设置连接的密码 + options.setPassword("password".toCharArray()); + // 设置超时时间 单位为秒 + options.setConnectionTimeout(10); + // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制 + options.setKeepAliveInterval(20); + //设置自动连接 + options.setAutomaticReconnect(true); + // 设置回调 + client.setCallback(new MqttCallbackExtended() { + //连接成功回调，需要重新订阅主题 + @Override + public void connectComplete(boolean reconnect, String serverURI) { + sendNotify(true); + subscribe(); + } + + @Override + public void connectionLost(Throwable cause) { + retryConnection(3); + } + + @Override + public void deliveryComplete(IMqttDeliveryToken token) { + token.isComplete(); + } + + @Override + public void messageArrived(String topic, MqttMessage message) { + } + }); + MqttTopic mqttTopic = client.getTopic("wyl"); + //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息 + options.setWill(mqttTopic, "close".getBytes(), 2, true); + client.connect(options); + subscribe(); + } catch (Exception e) { + log.error("mqtt异常" + e.getMessage()); + e.printStackTrace(); + } + } + + private void subscribe() { + try { + //订阅队列 + client.subscribe("wyl", 2, (topic, message) -> { + String msg = new String(message.getPayload(), StandardCharsets.UTF_8); + log.warn("mqtt收到信息:"+msg); + //处理消息 + }); + } catch (MqttException e) { + log.error("mqtt异常" + e.getMessage()); + } + } + + private void retryConnection(int retryNumber) { + //当重试多次，依旧失败，就是机器故障，需要通知人工处理 + if (retryNumber < 0) { + sendNotify(false); + } + try { + client.reconnect(); + TimeUnit.SECONDS.sleep(30); + if (!client.isConnected()) { + retryConnection(--retryNumber); + } + } catch (MqttException | InterruptedException e) { + log.error("mqtt异常:"+e.getMessage()); + } + } + + + //发送重试通知 + private void sendNotify(boolean isSuccess) { + if (isSuccess){ + log.debug("mqtt连接成功"); + }else { + log.error("mqtt连接失败"); + } + } + + + public static void main(String[] args) { + ClientMQTT2 clientMQTT = new ClientMQTT2(); + clientMQTT.subscribe(); + } +} \ No newline at end of file diff --git a/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT3.java b/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT3.java new file mode 100644 index 0000000..6efaef9 --- /dev/null +++ b/mqtt/src/main/java/com/wyl/mqtt/ClientMQTT3.java @@ -0,0 +1,111 @@ +package com.wyl.mqtt; + +import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.TimeUnit; + +@Slf4j +public class ClientMQTT3 { + MqttClient client; + MqttConnectOptions options; + + public ClientMQTT3() { + try { + // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存 + client = new MqttClient("tcp://127.0.0.1:61613", "wyl03", new MemoryPersistence()); + // MQTT的连接设置 + options = new MqttConnectOptions(); + // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接 + options.setCleanSession(true); + // 设置连接的用户名 + options.setUserName("admin"); + // 设置连接的密码 + options.setPassword("password".toCharArray()); + // 设置超时时间 单位为秒 + options.setConnectionTimeout(10); + // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制 + options.setKeepAliveInterval(20); + //设置自动连接 + options.setAutomaticReconnect(true); + // 设置回调 + client.setCallback(new MqttCallbackExtended() { + //连接成功回调，需要重新订阅主题 + @Override + public void connectComplete(boolean reconnect, String serverURI) { + sendNotify(true); + subscribe(); + } + + @Override + public void connectionLost(Throwable cause) { + retryConnection(3); + } + + @Override + public void deliveryComplete(IMqttDeliveryToken token) { + token.isComplete(); + } + + @Override + public void messageArrived(String topic, MqttMessage message) { + } + }); + MqttTopic mqttTopic = client.getTopic("wyl"); + //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息 + options.setWill(mqttTopic, "close".getBytes(), 2, true); + client.connect(options); + subscribe(); + } catch (Exception e) { + log.error("mqtt异常" + e.getMessage()); + e.printStackTrace(); + } + } + + private void subscribe() { + try { + //订阅队列 + client.subscribe("wyl", 2, (topic, message) -> { + String msg = new String(message.getPayload(), StandardCharsets.UTF_8); + log.warn("mqtt收到信息:"+msg); + //处理消息 + }); + } catch (MqttException e) { + log.error("mqtt异常" + e.getMessage()); + } + } + + private void retryConnection(int retryNumber) { + //当重试多次，依旧失败，就是机器故障，需要通知人工处理 + if (retryNumber < 0) { + sendNotify(false); + } + try { + client.reconnect(); + TimeUnit.SECONDS.sleep(30); + if (!client.isConnected()) { + retryConnection(--retryNumber); + } + } catch (MqttException | InterruptedException e) { + log.error("mqtt异常:"+e.getMessage()); + } + } + + + //发送重试通知 + private void sendNotify(boolean isSuccess) { + if (isSuccess){ + log.debug("mqtt连接成功"); + }else { + log.error("mqtt连接失败"); + } + } + + + public static void main(String[] args) { + ClientMQTT3 clientMQTT = new ClientMQTT3(); + clientMQTT.subscribe(); + } +} \ No newline at end of file diff --git a/mqtt/src/main/java/com/wyl/mqtt/MqttClientPublish.java b/mqtt/src/main/java/com/wyl/mqtt/MqttClientPublish.java new file mode 100644 index 0000000..d5d5124 --- /dev/null +++ b/mqtt/src/main/java/com/wyl/mqtt/MqttClientPublish.java @@ -0,0 +1,95 @@ +package com.wyl.mqtt; + +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.atomic.AtomicInteger; + +public class MqttClientPublish { + + private static AtomicInteger count = new AtomicInteger(); + private static int qos = 1; + private static String clientId = "JavaClientPublish"; + private static String broker = "tcp://127.0.0.1:1883"; + private static String topic = "wyl01"; + private static final Object lock = new Object(); + public static final int THREAD_COUNT = 2; + private static CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT); + + public static void main(String[] args) throws Exception { + MqttConnectOptions connOpts = new MqttConnectOptions(); + connOpts.setServerURIs(new String[] {broker}); + connOpts.setMaxInflight(THREAD_COUNT * 10); + connOpts.setCleanSession(false); + connOpts.setAutomaticReconnect(true); + connOpts.setKeepAliveInterval(30); + connOpts.setUserName("wyl"); + connOpts.setPassword("wyl".toCharArray()); + MemoryPersistence persistence = new MemoryPersistence(); + MqttAsyncClient client = new MqttAsyncClient(connOpts.getServerURIs()[0],clientId, persistence); + CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT); + + client.connect(connOpts,null,new IMqttActionListener(){ + //监听器快速返回控制非常重要，否则MQTT客户端的操作将会停止 + @Override + public void onSuccess(IMqttToken asyncActionToken) { + for (int i = 0; i < THREAD_COUNT; i++) { + new Sender(barrier, client).start(); + } + } + + @Override + public void onFailure(IMqttToken asyncActionToken, Throwable exception) { + exception.printStackTrace(); + } + }); + countDownLatch.await(); + + System.out.println("发布完毕!"); + + } + + private static class Sender extends Thread { + private CyclicBarrier barrier; + private MqttAsyncClient asyncClient; + + public Sender(CyclicBarrier barrier, MqttAsyncClient asyncClient) { + this.barrier = barrier; + this.asyncClient = asyncClient; + } + + @Override + public void run() { + try { + barrier.await(); + } catch (Exception e) { + e.printStackTrace(); + } + int index = count.incrementAndGet(); + String content = index + ""; + MqttMessage message = new MqttMessage(); + message.setQos(qos); + message.setPayload(content.getBytes()); + message.setId(index); + try { + asyncClient.publish(topic, message,null,new IMqttActionListener(){ + + @Override + public void onSuccess(IMqttToken asyncActionToken) { + countDownLatch.countDown(); + } + + @Override + public void onFailure(IMqttToken asyncActionToken, Throwable exception) { + + } + }); + } catch (Exception e) { + e.printStackTrace(); + } + } + } +} \ No newline at end of file diff --git a/mqtt/src/main/java/com/wyl/mqtt/MqttClientV1.java b/mqtt/src/main/java/com/wyl/mqtt/MqttClientV1.java new file mode 100644 index 0000000..0815a7a --- /dev/null +++ b/mqtt/src/main/java/com/wyl/mqtt/MqttClientV1.java @@ -0,0 +1,190 @@ +package com.wyl.mqtt; + +import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.client.mqttv3.MqttConnectOptions; +import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; +import org.eclipse.paho.client.mqttv3.MqttException; +import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.eclipse.paho.client.mqttv3.MqttPersistenceException; +import org.eclipse.paho.client.mqttv3.MqttTopic; +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; + +@Slf4j +public class MqttClientV1 { + + public static org.eclipse.paho.client.mqttv3.MqttClient mqttClient = null; + private static MemoryPersistence memoryPersistence = null; + private static MqttConnectOptions mqttConnectOptions = null; + + private static MqttClientV1 instance = null; + + public static MqttClientV1 getInstance() throws Exception { + if (instance == null) { + synchronized (MqttClientV1.class) { + if (instance == null) { + instance = new MqttClientV1(); + } + } + } + return instance; + } + + public MqttClientV1(){ + init("admin"); + } + + public void init(String clientId) { + //初始化连接设置对象 + mqttConnectOptions = new MqttConnectOptions(); + //初始化MqttClient + if(null != mqttConnectOptions) { +// true可以安全地使用内存持久性作为客户端断开连接时清除的所有状态 + mqttConnectOptions.setCleanSession(true); +// 设置连接超时 + mqttConnectOptions.setConnectionTimeout(30); + mqttConnectOptions.setUserName("admin"); + mqttConnectOptions.setPassword("password".toCharArray()); +// 设置持久化方式 + memoryPersistence = new MemoryPersistence(); + if(null != memoryPersistence && null != clientId) { + try { + mqttClient = new org.eclipse.paho.client.mqttv3.MqttClient("tcp://127.0.0.1:61613", clientId,memoryPersistence); + } catch (MqttException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + }else { + + } + }else { + log.error("mqttConnectOptions对象为空"); + } + //设置连接和回调 + if(null != mqttClient) { + if(!mqttClient.isConnected()) { + try { + log.info("创建连接:" + mqttClient.isConnected()); + mqttClient.connect(mqttConnectOptions); + } catch (MqttException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + + } + }else { + log.error("mqttClient为空"); + } + } + + // 关闭连接 + public void closeConnect() { + //关闭存储方式 + if(null != memoryPersistence) { + try { + memoryPersistence.close(); + } catch (MqttPersistenceException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + }else { + log.error("memoryPersistence is null"); + } + +// 关闭连接 + if(null != mqttClient) { + if(mqttClient.isConnected()) { + try { + mqttClient.disconnect(); + mqttClient.close(); + } catch (MqttException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + }else { + log.error("mqttClient is not connect"); + } + }else { + log.error("mqttClient is null"); + } + } + + // 发布消息 + public void publishMessage(String pubTopic,String message,int qos) { + if(null != mqttClient&& mqttClient.isConnected()) { + MqttMessage mqttMessage = new MqttMessage(); + mqttMessage.setQos(qos); + mqttMessage.setPayload(message.getBytes()); + MqttTopic topic = mqttClient.getTopic(pubTopic); + if(null != topic) { + try { + MqttDeliveryToken publish = topic.publish(mqttMessage); + if(!publish.isComplete()) { + //log.info("消息发布成功"); + } + } catch (MqttException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + + }else { + reConnect(); + } + + } + // 重新连接 + public void reConnect() { + if(null != mqttClient) { + if(!mqttClient.isConnected()) { + if(null != mqttConnectOptions) { + try { + mqttClient.connect(mqttConnectOptions); + } catch (MqttException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + }else { + log.error("mqttConnectOptions is null"); + } + }else { + log.error("mqttClient is null or connect"); + } + }else { + init("admin"); + } + + } + // 订阅主题 + public void subTopic(String topic) { + if(null != mqttClient&& mqttClient.isConnected()) { + try { + mqttClient.subscribe(topic, 1); + } catch (MqttException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + }else { + log.error("mqttClient is error"); + } + } + + + // 清空主题 + public void cleanTopic(String topic) { + if(null != mqttClient&& !mqttClient.isConnected()) { + try { + mqttClient.unsubscribe(topic); + } catch (MqttException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + }else { + log.error("mqttClient is error"); + } + } + + public static void main(String [] args){ + MqttClientV1 mqttClient = new MqttClientV1(); + mqttClient.publishMessage("wyl", "12312312312", 1); + } +} \ No newline at end of file diff --git a/mqtt/src/main/java/com/wyl/mqtt/SimpleMqttClient.java b/mqtt/src/main/java/com/wyl/mqtt/SimpleMqttClient.java new file mode 100644 index 0000000..a5d85a1 --- /dev/null +++ b/mqtt/src/main/java/com/wyl/mqtt/SimpleMqttClient.java @@ -0,0 +1,146 @@ +package com.wyl.mqtt; + +import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; + +@Slf4j +public class SimpleMqttClient { + + //全局唯一 单例 + private static IMqttAsyncClient client; + + private static IMqttAsyncClient getClient() { + return client; + } + private static void setClient(IMqttAsyncClient client) { + SimpleMqttClient.client = client; + } + + /** + * 连接MQTT服务器 + */ + public void connect(String serverURI, String clientID, String username, String password) { + + IMqttAsyncClient client = null; + try { + client = new MqttAsyncClient(serverURI, clientID, new MemoryPersistence()); + + MqttConnectOptions options = new MqttConnectOptions(); + options.setCleanSession(false); + options.setUserName(username); + options.setPassword(password.toCharArray()); + options.setServerURIs(new String[]{serverURI}); + options.setConnectionTimeout(100); + options.setAutomaticReconnect(true); + //设置心跳 + options.setKeepAliveInterval(30); + client.setCallback(new MqttCallbackExtended() { + @Override + public void connectComplete(boolean b, String s) { + log.info("重连成功！"); + //publish("BUS_TS_REPLY_TOPIC_湘AB7182", "7E81074000010000000060013608025900000F7E", 0, false); + + } + + + @Override + public void connectionLost(Throwable throwable) { + log.error("Lost connection!!! {}"); + throwable.printStackTrace(); + } + + @Override + public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { + log.info("接收消息主题 : " + topic); + log.info("接收消息Qos : " + mqttMessage.getQos()); + } + + + @Override + public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { + log.debug("send success ? --> {}, {}", iMqttDeliveryToken.isComplete(), 1); + } + }); + + + try { + client.connect(options); + SimpleMqttClient.setClient(client); + } catch (Exception e) { + e.printStackTrace(); + } + + } catch (Exception e) { + e.printStackTrace(); + } + } + + /** + * 发布 + * + * @param qos 连接方式 + * @param retained 是否保留 + * @param topic 主题 + * @param pushMessage 消息体 + */ + public void publish(String topic, byte[] message, int qos, boolean retained) { + + if(client != null && client.isConnected()) { + try { + IMqttDeliveryToken token = client.publish(topic, message, qos, retained); + token.waitForCompletion(); + log.debug("Is the message sent successfully? --> {}, {}", token.isComplete()); + } catch (MqttException e) { + e.printStackTrace(); + } + } + } + + /** + * 订阅某个主题 + * + * @param topic 主题 + * @param qos 连接方式 + */ + public void subscribe(String topic, int qos) { + log.info("开始订阅主题: {}" , topic); + if (client != null && client.isConnected()) { + try { + IMqttToken token = client.subscribe(topic, qos); + //token.waitForCompletion(); + } catch (MqttException e) { + e.printStackTrace(); + log.error("subscribe topic {} qos {} error!", topic, qos); + } + } + } + /** + * 订阅多主题 + * + * @param topic 主题 + * @param qos 连接方式 + */ + public void subscribe(String[] topics, int[] qos) { + log.info("开始订阅主题集合:{}", Arrays.asList(topics)); + if (client != null && client.isConnected()) { + try { + IMqttToken token = client.subscribe(topics, qos); + //token.waitForCompletion(); + } catch (MqttException e) { + e.printStackTrace(); + log.error("subscribe topic {} qos {} error!", topics, qos); + } + } + } + + public static void main(String[] args) { + SimpleMqttClient simpleMqttClient = new SimpleMqttClient(); + simpleMqttClient.connect("tcp://127.0.0.1:1883", "123", "wyl","wyl"); + simpleMqttClient.publish("wyl001", "123".getBytes(StandardCharsets.UTF_8), 0, false); + } +} + diff --git a/pom.xml b/pom.xml index 02d4715..c7c053d 100644 --- a/pom.xml +++ b/pom.xml @@ -17,6 +17,8 @@ spring-boot redis DelayQueue + javacv + mqtt pom @@ -39,7 +41,7 @@ org.apache.kafka kafka-clients - 2.8.1 + 3.4.0 com.xuxueli