JavaSwing简单发送/接收kafka数据，并打包成exe（3）

ゝ一纸荒年。 2022-05-27 01:38 48阅读 0赞

上一篇链接[https://blog.csdn.net/rico\_zhou/article/details/79865102][https_blog.csdn.net_rico_zhou_article_details_79865102]

本文实现具体的发送，消费kafka数据代码。

kafka服务的搭建详见另一篇文章。

[https://blog.csdn.net/rico\_zhou/article/details/79866076][https_blog.csdn.net_rico_zhou_article_details_79866076]

kafka服务ip为192.168.1.140，端口port为默认9092。

上一篇文章前台代码基本完成，接下来我们在kafka包内接着撸。

创建类SendMsgKafka，创建一个全局静态变量显示发送的次数

public static int TIMES = 0;

接下来就是具体的方法，先写发送数据的代码，创建一个总体控制的方法kafkaController(BaseMsg baseMsg)

// 处理参数
    	public ReturnMsg kafkaController(BaseMsg baseMsg) throws Exception {
    		// 实例化一个封装参数的对象
    		kafkaMsg kafkaMsg = new kafkaMsg();
    		// 转化前台后台参数
    		handleParams(baseMsg, kafkaMsg);
    
    		// send
    		ReturnMsg returnMsg = sendMsg(kafkaMsg);
    		// 返回信息
    		return returnMsg;
    
    	}

前后台参数转化方法

// 前后台参数转化
    	private void handleParams(BaseMsg baseMsg, kafkaMsg kafkaMsg) {
    		// ip和port组合到一起
    		kafkaMsg.setIpPort(baseMsg.getKafkaIp() + ":" + baseMsg.getKafkaPort());
    		kafkaMsg.setGroupId(baseMsg.getKafkaGroupId());
    		kafkaMsg.setTopic(baseMsg.getKafkaTopic());
    		kafkaMsg.setKey(baseMsg.getKafkaKey());
    		// 条数数据类型转化
    		kafkaMsg.setAccount(Integer.valueOf(baseMsg.getKafkaAccount()));
    		// 间隔时间数据类型转化
    		kafkaMsg.setTime(Integer.valueOf(baseMsg.getKafkaTimes()));
    		// 起始id数据类型转化
    		kafkaMsg.setStartId(Integer.valueOf(baseMsg.getKafkaStartId()));
    		// 若是发送默认信息，json格式
    		if (baseMsg.isKafkaIsDefault()) {
    			String msgJson = "{\"name\":\"testName\",\"age\":1,\"address\":\"address\",\"otherMsg\":\"msgtest\"}";
    			kafkaMsg.setMessage(msgJson);
    		} else {
    			kafkaMsg.setMessage(baseMsg.getKafkaMessage());
    		}
    	}

编写发送数据的核心方法sendMsg(kafkaMsg kafkaMsg)

// send
    	public ReturnMsg sendMsg(kafkaMsg kafkaMsg) throws IOException, Exception {
    		PrintStream out = null;
    		PrintStream mytxt = null;
    		// 时间戳
    		try {
    			SimpleDateFormat sd = new SimpleDateFormat("yyyyMMddhhmmss");
    			mytxt = new PrintStream("./kafkalog" + sd.format(new Date()) + ".txt");
    			// 拦截输出流，将重要信息输出成文件保存到本地
    			out = System.out;
    			System.setOut(mytxt);
    			System.out.println("Start：");
    
    		} catch (FileNotFoundException e) {
    			e.printStackTrace();
    		}
    		// 发送次数初始化
    		TIMES = 0;
    		// kafka参数配置
    		Properties configProperties = new Properties();
    		configProperties.put("group.id", kafkaMsg.getGroupId());
    		configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaMsg.getIpPort());
    		configProperties.put("key.serializer", StringSerializer.class);
    		configProperties.put("value.serializer", JsonSerializer.class);
    
    		KafkaProducer producer = new KafkaProducer(configProperties);
    
    		ObjectMapper objectMapper = new ObjectMapper();
    		int i = 0;
    		int acc = kafkaMsg.getAccount();
    		while (i < acc) {
    			JSONObject data = new JSONObject();
    			// 对所需要发送的数据进行预处理
    			// handlemessage
    			String msg = hanleMsg(kafkaMsg, i, acc);
    			// 转为json发送
    			JsonNode jsonNode = objectMapper.readTree(msg);
    			ProducerRecord rec = new ProducerRecord(kafkaMsg.getTopic(), kafkaMsg.getKey() + i, jsonNode);
    			producer.send(rec);
    			System.out.println(jsonNode);
    			Thread.sleep(kafkaMsg.getTime());
    			i++;
    			TIMES++;
    			rm.setComTimes(TIMES);
    			// 线程状态判断
    			if (!KafkaMainGUI.isRunning) {
    				rm.setCompleted(false);
    				System.setOut(out);
    				mytxt.close();
    				producer.close();
    				return rm;
    			}
    		}
    		rm.setCompleted(true);
    		System.out.println("End");
    		System.setOut(out);
    		mytxt.close();
    		// 注意：一定要将kafka连接关闭
    		producer.close();
    		return rm;
    
    	}

这个方法有一些注意点，一是将重要的发送数据生成文件保存到本地，方便查看，二是要对数据进行预处理，因为发送多条一样的数据没有意义，所以将数据处理下，三是对线程状态进行判断，以便于中止按钮对后台程序产生效果，四是一定要将kafka连接close掉，否则会出现错误。

下面编写一个处理数据的方法

// handlemsg
    	private String hanleMsg(kafkaMsg kafkaMsg, int i, int acc) throws IOException {
    		com.alibaba.fastjson.JSONObject json = JSON.parseObject(kafkaMsg.getMessage());
    		// 只发送一条信息则不做更改
    		if (acc != 1) {
    			for (Entry<String, Object> entry : json.entrySet()) {
    				if ("age".equals(entry.getKey())) {
    					int age = Integer.valueOf(entry.getValue().toString()) + i;
    					entry.setValue(age);
    				} else {
    					entry.setValue(entry.getValue().toString() + i);
    				}
    			}
    		}
    		String s = json.toString();
    		int idn = i + kafkaMsg.getStartId();
    		s = "{" + "\"id\":" + idn + "," + s.substring(1);
    		return s;
    	}

基本逻辑就是如果发送次数只有一条则不做更改，仅添加一个字段id，如果次数大于1，则将每个字段值后缀都顺序增加，注意本程序只是简单的测试发送消费kafka数据功能，实际中不需要这样更改数据，只是为了测试而已。

接下来书写消费数据的代码，上一篇文章介绍已经把主要逻辑放在前台处理了，后台只处理下配置信息

public Properties kafkaController2(BaseMsg baseMsg) {
    		Properties props = new Properties();
    		props.put("bootstrap.servers", baseMsg.getKafkaIp() + ":" + baseMsg.getKafkaPort());
    		props.put("group.id", baseMsg.getKafkaGroupId());
    		props.put("enable.auto.commit", "true");
    		props.put("auto.commit.interval.ms", "1000");
    		props.put("auto.offset.reset", "earliest");
    		props.put("session.timeout.ms", "30000");
    		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    		return props;
    	}

到此为止，基本上程序完成了，并且可以实现发送消费数据了，下一篇我们再添加一些简单功能以便于更加人性化。

GitHub源码：

[https://github.com/ricozhou/msgkafka][https_github.com_ricozhou_msgkafka]

[https_blog.csdn.net_rico_zhou_article_details_79865102]: https://blog.csdn.net/rico_zhou/article/details/79865102
[https_blog.csdn.net_rico_zhou_article_details_79866076]: https://blog.csdn.net/rico_zhou/article/details/79866076
[https_github.com_ricozhou_msgkafka]: https://github.com/ricozhou/msgkafka