five-groups
/
five-groups-couplet
4
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases Wiki Activity

优化判断语句

server_five_dongxiaodong
dongxiaodong 2024-04-09 09:57:28 +08:00
parent 9f17578564
commit 54a091e3b6
2 changed files with 84 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java
View File

@ -76,11 +76,10 @@ public class ModelsKafkaMessage {
* *
* @return * @return
*/ */
@Scheduled(fixedDelay = 50) // @Scheduled(fixedDelay = 50)
@PostConstruct
public void consumerMessages() { public void consumerMessages() {
executorService.execute(this::consumer); executorService.execute(this::consumer);
} }
public void consumer() { public void consumer() {

164
couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/controller/KafkaTest.java
View File

@ -1,93 +1,93 @@
package com.couplet.mq.controller; //package com.couplet.mq.controller;
//
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig; //import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords; //import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer; //import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer; //import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig; //import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord; //import org.apache.kafka.clients.producer.ProducerRecord;
//
import javax.annotation.PostConstruct; //import javax.annotation.PostConstruct;
import java.time.Duration; //import java.time.Duration;
import java.util.Collections; //import java.util.Collections;
import java.util.Properties; //import java.util.Properties;
//
/** ///**
* @ProjectName: five-groups-couplet // * @ProjectName: five-groups-couplet
* @Author: LiuYunHu // * @Author: LiuYunHu
* @CreateTime: 2024/4/4 // * @CreateTime: 2024/4/4
* @Description: kafka测试类 // * @Description: kafka测试类
*/ // */
//
@Slf4j //@Slf4j
public class KafkaTest { //public class KafkaTest {
private static final String TOPIC_NAME = "online"; // private static final String TOPIC_NAME = "online";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092"; // private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
//
public static void main(String[] args) { // public static void main(String[] args) {
//生产者示例 // //生产者示例
// produceMessage(); //// produceMessage();
//
//消费者示例 // //消费者示例
// consumerMessages(); //// consumerMessages();
//
} // }
//
//生产者 // //生产者
@PostConstruct // @PostConstruct
private static void produceMessage() { // private static void produceMessage() {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
//创建生产者
try {
//发送消息
for (int i = 0; i < 10000; i++) {
String message = "佳佳来喽" + (i + 1);
producer.send(new ProducerRecord<>(TOPIC_NAME, message));
System.out.println("发送消息：" + message);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
producer.close();
}
}
//消费者
// private static void consumerMessages() {
// Properties props = new Properties(); // Properties props = new Properties();
// props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); // props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
// props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); // props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
// props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); // props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
// props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//
// //创建消费者
// KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// //
// KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// //创建生产者
// try { // try {
// //
// //订阅主题 // //发送消息
// consumer.subscribe(Collections.singletonList(TOPIC_NAME)); // for (int i = 0; i < 10000; i++) {
// String message = "佳佳来喽" + (i + 1);
// producer.send(new ProducerRecord<>(TOPIC_NAME, message));
// //
// //持续消费消息 // System.out.println("发送消息：" + message);
// while (true) {
// ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// records.forEach(record -> {
// System.out.println("消费者接受到的消息值：" + record.value());
// });
// } // }
// } catch (Exception e) { // } catch (Exception e) {
// e.printStackTrace(); // e.printStackTrace();
// } finally { // } finally {
// consumer.close(); // producer.close();
// } // }
// } // }
//
} // //消费者
//// private static void consumerMessages() {
//// Properties props = new Properties();
//// props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
//// props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
//// props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//// props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
////
//// //创建消费者
//// KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
////
//// try {
////
//// //订阅主题
//// consumer.subscribe(Collections.singletonList(TOPIC_NAME));
////
//// //持续消费消息
//// while (true) {
//// ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
//// records.forEach(record -> {
//// System.out.println("消费者接受到的消息值：" + record.value());
//// });
//// }
//// } catch (Exception e) {
//// e.printStackTrace();
//// } finally {
//// consumer.close();
//// }
//// }
//
//}