From 8e05fd0d300d9857bdb1eacac9c260178ef18833 Mon Sep 17 00:00:00 2001
From: chenbingxuan <3581044601@qq.com>
Date: Mon, 11 Dec 2023 09:28:33 +0800
Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=BA=90?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../service/impl/DataDatabaseServiceImpl.java | 31 +++++++++++--------
 1 file changed, 18 insertions(+), 13 deletions(-)

diff --git a/srt-cloud-data-integrate/src/main/java/net/srt/service/impl/DataDatabaseServiceImpl.java b/srt-cloud-data-integrate/src/main/java/net/srt/service/impl/DataDatabaseServiceImpl.java
index 8ef19b6..a3bc3e9 100644
--- a/srt-cloud-data-integrate/src/main/java/net/srt/service/impl/DataDatabaseServiceImpl.java
+++ b/srt-cloud-data-integrate/src/main/java/net/srt/service/impl/DataDatabaseServiceImpl.java
@@ -161,7 +161,7 @@ public class DataDatabaseServiceImpl extends BaseServiceImpl<DataDatabaseDao, Da
 			);
 		}else {
 			//判断是什么类型进行连接
-			if(vo.getDatabaseType()== Judgment.REDIS.getIndex()){
+			if(vo.getDatabaseType().equals(Judgment.REDIS.getIndex())){
 				//连接本地的 Redis 服务
 				// 创建 Jedis 对象
 				Jedis jedis = new Jedis(vo.getDatabaseIp(), Integer.valueOf(vo.getDatabasePort()));
@@ -184,28 +184,33 @@ public class DataDatabaseServiceImpl extends BaseServiceImpl<DataDatabaseDao, Da
 				}
 			}
 
-			if(vo.getDatabaseType()==Judgment.KAFKA.getIndex()){
+			if(vo.getDatabaseType().equals(Judgment.KAFKA.getIndex())){
 				// 配置生产者属性
 				Properties properties = new Properties();
-				String ip=vo.getDatabaseIp()+":"+vo.getDatabasePort();
+				String ip = "invalid_ip:invalid_port"; // 设置一个无效的IP和端口，以模拟连接失败的情况
 				properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ip);
 				properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class.getName());
 				properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class.getName());
 
 				// 创建生产者
-				Producer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<>(properties);
+				Producer<String, String> producer = null;
+				try {
+					producer = new org.apache.kafka.clients.producer.KafkaProducer<>(properties);
 
-				// 消息
-				String message = "Default Message";
+					// 消息
+					String message = "Default Message";
 
-				// 发送消息到主题
-                try {
-                    producer.send(new ProducerRecord<String, String>(vo.getDatabaseKafKa(), message));
-                    System.out.println("消息发送成功");
-                } finally {
+					// 发送消息到主题
+					producer.send(new ProducerRecord<>(vo.getDatabaseKafKa(), message));
+					System.out.println("消息发送成功");
+				} catch (Exception e) {
+					System.err.println("连接失败: " + e.getMessage());
+				} finally {
 					// 关闭生产者
-					producer.close();
-                }
+					if (producer != null) {
+						producer.close();
+					}
+				}
 			}
 		}
 		if (vo.getId() != null) {