wyl/JavaBasiceDemo
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

test

Browse Source
This commit is contained in:
wyl
2023-04-23 17:28:54 +08:00
parent 760582618f
commit 73d5c27005
19 changed files with 1033 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
example_output.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 34 KiB

file-read/src/main/java/com/wyl/file/sequence/Sequential.java
+4 -3
View File
@@ -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();
}
file-read/src/test/java/com/wyl/file/FileTest.java
+6 -7
View File
@@ -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();
javacv/pom.xml
+26
View File
@@ -0,0 +1,26 @@
<?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">
<parent>
<artifactId>JavaBasiceDemo</artifactId>
<groupId>com.wyl.example</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>javacv</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.bytedeco</groupId>
<artifactId>javacv-platform</artifactId>
<version>1.5.8</version>
</dependency>
</dependencies>
</project>
javacv/src/main/java/com/wyl/javacv/ImageToVideoExample.java
+52
View File
@@ -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("视频合成完成！");
}
}
javacv/src/main/java/com/wyl/javacv/JavaCVExample.java
+33
View File
@@ -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();
}
}
javacv/src/main/resources/OIP.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 17 KiB

javacv/src/main/resources/R.jpeg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 730 KiB

javacv/src/main/resources/example_output.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 34 KiB

kafka/src/test/java/com/wyl/kafka/test/ConsumerTest.java
+42
View File
@@ -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<String, String> consumer = new KafkaConsumer<>(consumerProps);
// 接收图片消息从 Kafka
consumer.subscribe(Arrays.asList("wyl-events"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> 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);
}
}
}
}
kafka/src/test/java/com/wyl/kafka/test/ProducerTest.java
+79
View File
@@ -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<String, String> 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<String, String> 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<String, String> 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));
}
}
}
mqtt/pom.xml
+27
View File
@@ -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">
<parent>
<artifactId>JavaBasiceDemo</artifactId>
<groupId>com.wyl.example</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>mqtt</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<!-- https://mvnrepository.com/artifact/org.eclipse.paho/org.eclipse.paho.client.mqttv3 -->
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
</dependencies>
</project>
mqtt/src/main/java/com/wyl/mqtt/ClientMQTT.java
+108
View File
@@ -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();
}
}
mqtt/src/main/java/com/wyl/mqtt/ClientMQTT2.java
+111
View File
@@ -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();
}
}
mqtt/src/main/java/com/wyl/mqtt/ClientMQTT3.java
+111
View File
@@ -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();
}
}
mqtt/src/main/java/com/wyl/mqtt/MqttClientPublish.java
+95
View File
@@ -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();
}
}
}
}
mqtt/src/main/java/com/wyl/mqtt/MqttClientV1.java
+190
View File
@@ -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);
}
}
mqtt/src/main/java/com/wyl/mqtt/SimpleMqttClient.java
+146
View File
@@ -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);
}
}
pom.xml
+3 -1
View File
@@ -17,6 +17,8 @@
<module>spring-boot</module>
<module>redis</module>
<module>DelayQueue</module>
<module>javacv</module>
<module>mqtt</module>
</modules>
<packaging>pom</packaging>
<properties>
@@ -39,7 +41,7 @@
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.1</version>
<version>3.4.0</version>
</dependency>
<dependency>
<groupId>com.xuxueli</groupId>