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

不念不忘少年蓝@ 2022-05-27 01:23 45阅读 0赞

上一篇文章链接[https://blog.csdn.net/rico\_zhou/article/details/79857625][https_blog.csdn.net_rico_zhou_article_details_79857625]

这篇主要实现发送数据到后台功能，后台发送数据功能见下一篇文章，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。

上一篇基本GUI界面已实现，现在开始书写监听代码。

还是类KafkaMainGUI，先写几个方法，获取前台参数，校验前台参数

获取前台参数

// get base set
    	private BaseMsg getBaseMsg() {
    		BaseMsg baseMsg = new BaseMsg();
    		baseMsg.setKafkaIp(tt1.getText().trim());
    		baseMsg.setKafkaPort(tt2.getText().trim());
    		baseMsg.setKafkaGroupId(tt3.getText().trim());
    		baseMsg.setKafkaTopic(tt4.getText().trim());
    		baseMsg.setKafkaAccount(tt5.getText().trim());
    		baseMsg.setKafkaTimes(tt6.getText().trim());
    		baseMsg.setKafkaKey(tt7.getText().trim());
    		baseMsg.setKafkaIsDefault(jrb1.isSelected());
    		baseMsg.setKafkaMessage(jta1.getText().trim());
    		baseMsg.setKafkaStartId(tt8.getText().trim());
    		return baseMsg;
    	}

校验前台参数

private int checkParams(BaseMsg baseMsg) {
    		if (!utils.checkBase1(baseMsg, 0)) {
    			return 1;
    		}
    		if (!utils.checkBase2(baseMsg, 0)) {
    			return 2;
    		}
    		if (!utils.checkIp(baseMsg.getKafkaIp())) {
    			return 3;
    		}
    		if (!utils.checkAccTimeId(baseMsg)) {
    			return 4;
    		}
    		if (!baseMsg.isKafkaIsDefault()) {
    			if (!utils.checkMsg(baseMsg.getKafkaMessage())) {
    				return 5;
    			}
    		}
    
    		return 0;
    	}

在Utils类中写各种具体的检验方法

// basecheck
    	public boolean checkBase1(BaseMsg baseMsg, int i) {
    		// not null ""
    		if ("".equals(baseMsg.getKafkaIp()) || "".equals(baseMsg.getKafkaPort()) || "".equals(baseMsg.getKafkaGroupId())
    				|| "".equals(baseMsg.getKafkaTopic())) {
    			return false;
    		}
    		if (i == 0) {
    			if ("".equals(baseMsg.getKafkaAccount()) || "".equals(baseMsg.getKafkaTimes())
    					|| "".equals(baseMsg.getKafkaKey()) || "".equals(baseMsg.getKafkaStartId())) {
    				return false;
    			}
    		}
    
    		return true;
    	}
    
    	public boolean checkBase2(BaseMsg baseMsg, int i) {
    		// only number
    		if (!baseMsg.getKafkaPort().matches("^[0-9]*$")) {
    			return false;
    		}
    		if (i == 0) {
    			if (!baseMsg.getKafkaAccount().matches("^[0-9]*$") || !baseMsg.getKafkaTimes().matches("^[0-9]*$")
    					|| !baseMsg.getKafkaStartId().matches("^[0-9]*$")) {
    				return false;
    			}
    		}
    		return true;
    	}
    
    	// checkip
    	public boolean checkIp(String ip) {
    		// 定义正则表达式
    		String regex = "^(1\\d{2}|2[0-4]\\d|25[0-5]|[1-9]\\d|[1-9])\\." + "(1\\d{2}|2[0-4]\\d|25[0-5]|[1-9]\\d|\\d)\\."
    				+ "(1\\d{2}|2[0-4]\\d|25[0-5]|[1-9]\\d|\\d)\\." + "(1\\d{2}|2[0-4]\\d|25[0-5]|[1-9]\\d|\\d)$";
    		if (ip.matches(regex)) {
    			return true;
    		} else {
    			return false;
    		}
    	}
    
    	public boolean checkAccTimeId(BaseMsg baseMsg) {
    		if (baseMsg.getKafkaAccount().startsWith("0") || baseMsg.getKafkaTimes().startsWith("0")) {
    			return false;
    		}
    		if (baseMsg.getKafkaAccount().length() > 9 || baseMsg.getKafkaTimes().length() > 9
    				|| baseMsg.getKafkaStartId().length() > 9) {
    			return false;
    		}
    		return true;
    	}
    
    	public boolean checkMsg(String kafkaMessage) {
    		if ("".equals(kafkaMessage)) {
    			return false;
    		}
    		if (!checkJson(kafkaMessage)) {
    			return false;
    		}
    		return true;
    	}
    
    	private boolean checkJson(String kafkaMessage) {
    		try {
    			JSONObject jsonStr = JSONObject.parseObject(kafkaMessage);
    			return true;
    		} catch (Exception e) {
    			return false;
    		}
    	}

接下来监听开始按钮，KafkaMainGUI类

@Override
    	public void actionPerformed(ActionEvent e) {
    		// 监听开始
    		if (e.getSource().equals(button1)) {
    			// get params
    			baseMsg = getBaseMsg();
    			// check
    			int flag = checkParams(baseMsg);
    			if (flag == 0) {
    				button1.setEnabled(false);
    				button2.setEnabled(true);
    				button3.setEnabled(false);
    				jpbProcessLoading1.setVisible(true);
    				// jpbProcessLoading1.setIndeterminate(true); //
    				// 设置进度条为不确定模式,默认为确定模式
    				jpbProcessLoading1.setString("正在发送，请耐心等待...");
    				jpbProcessLoading1.setValue(0);
    
    				// start times
    				KafkaMainGUI.isRunning = true;
    				thread = new CountThread();
    				thread.start();
    
    				sw1 = new SwingWorker<ReturnMsg, String>() {
    					// 此方法是耗时任务
    					@Override
    					protected ReturnMsg doInBackground() throws Exception {
    						ReturnMsg returnMsg = null;
    						returnMsg = smk.kafkaController(baseMsg);
    						return returnMsg;
    					}
    
    					// done
    					protected void done() {
    						ReturnMsg returnMsg = null;
    						try {
    							returnMsg = get();
    						} catch (Exception e) {
    							e.printStackTrace();
    						}
    
    						thread.stop();
    						// jpbProcessLoading1.setIndeterminate(false);
    						if (returnMsg != null) {
    							if (returnMsg.isCompleted) {
    								jpbProcessLoading1.setString("发送完成！");
    								JOptionPane.showMessageDialog(null, "发送完成！共" + returnMsg.getComTimes() + "条", "提示消息",
    										JOptionPane.WARNING_MESSAGE);
    							} else {
    								jpbProcessLoading1.setString("发送中止！");
    								JOptionPane.showMessageDialog(null, "发送中止！共" + returnMsg.getComTimes() + "条", "提示消息",
    										JOptionPane.WARNING_MESSAGE);
    							}
    						} else {
    							jpbProcessLoading1.setString("发送失败！");
    							JOptionPane.showMessageDialog(null, "发送失败！请检查！", "提示消息", JOptionPane.WARNING_MESSAGE);
    						}
    						button1.setEnabled(true);
    						button2.setEnabled(false);
    						button3.setEnabled(true);
    					}
    				};
    				sw1.execute();
    			} else if (flag == 1) {
    				JOptionPane.showMessageDialog(null, "不能为空！请重新填写！", "提示消息", JOptionPane.WARNING_MESSAGE);
    			} else if (flag == 2) {
    				JOptionPane.showMessageDialog(null, "相关参数只能是数字！请重新填写！", "提示消息", JOptionPane.WARNING_MESSAGE);
    			} else if (flag == 3) {
    				JOptionPane.showMessageDialog(null, "IP不正确！请重新填写！", "提示消息", JOptionPane.WARNING_MESSAGE);
    			} else if (flag == 4) {
    				JOptionPane.showMessageDialog(null, "条数，间隔时间，初始ID超过范围！请重新填写！", "提示消息", JOptionPane.WARNING_MESSAGE);
    			} else if (flag == 5) {
    				JOptionPane.showMessageDialog(null, "Message格式不正确！请重新填写！", "提示消息", JOptionPane.WARNING_MESSAGE);
    			}
    
    		}
    		
    	}

这里简单介绍下SwingWorker线程，此线程安全，专门处理swing后台耗时任务，这样前台依然可以操作，但此线程不容易中止，所以设置一个全局静态变量isRunning来控制线程。

此线程有三个方法：

doInBackground()：专门处理耗时任务，启动线程开始调用此方法。

done()：此方法是后台耗时任务完成后调用接收返回值，get()即可获取。

process()：此方法可以在任务中动态的接收返回值。

再另起一个线程实时显示获取条数

// 实时显示发送的条数
    	private class CountThread extends Thread {
    		private CountThread() {
    			setDaemon(true);
    		}
    
    		@Override
    		public void run() {
    			jlb11.setVisible(true);
    			while (KafkaMainGUI.isRunning) {
    				jlb11.setText("Times:" + SendMsgKafka.TIMES);
    				jpbProcessLoading1.setValue(100 * SendMsgKafka.TIMES / Integer.valueOf(baseMsg.getKafkaAccount()));
    			}
    		}
    
    	}

同理，我们写消费数据的代码，在方法actionPerformed(ActionEvent e)中添加如下代码

// 监听接收
    		if (e.getSource().equals(button4)) {
    			// get params
    			baseMsg2 = getBaseMsg2();
    			// check
    			int flag = checkParams2(baseMsg2);
    			if (flag == 0) {
    				button4.setEnabled(false);
    				button5.setEnabled(true);
    				button6.setEnabled(false);
    				jpbProcessLoading2.setVisible(true);
    				jpbProcessLoading2.setIndeterminate(true); // 设置进度条为不确定模式,默认为确定模式
    				jpbProcessLoading2.setString("正在接收，请耐心等待...");
    				// start times
    				KafkaMainGUI.isRunning2 = true;
    
    				sw2 = new SwingWorker<String, String>() {
    					int times2 = 0;
    
    					// 此方法是耗时任务
    					@Override
    					protected String doInBackground() throws Exception {
    						Properties props = smk.kafkaController2(baseMsg2);
    						// smk.kafkaController3(baseMsg2);
    						KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
    						kafkaConsumer.subscribe(Arrays.asList(baseMsg2.getKafkaTopic()));
    						while (true) {
    							ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
    							for (ConsumerRecord<String, String> record : records) {
    								System.out.println("topic: " + record.topic() + " key: " + record.key() + " value: "
    										+ record.value() + " partition: " + record.partition());
    								// 实时发送到前台
    								publish("topic: " + record.topic() + " key: " + record.key() + " value: "
    										+ record.value() + " partition: " + record.partition());
    							}
    							if (!KafkaMainGUI.isRunning2) {
    								// close connection
    								kafkaConsumer.close();
    								return "1";
    							}
    						}
    					}
    
    					// done
    					protected void done() {
    						String f = "0";
    						try {
    							f = get();
    						} catch (Exception e) {
    							e.printStackTrace();
    						}
    						if ("1".equals(f)) {
    							// thread2.stop();
    							jpbProcessLoading2.setIndeterminate(false);
    							jpbProcessLoading2.setString("中止接收！");
    							button4.setEnabled(true);
    							button5.setEnabled(false);
    							button6.setEnabled(true);
    						}
    					}
    
    					@Override
    					protected void process(List<String> msgs) {
    						if (SwingUtilities.isEventDispatchThread()) {
    							jlb16.setVisible(true);
    							for (String msg : msgs) {
    								times2++;
    								jlb16.setText("Times:" + times2);
    								jta2.append(msg.toString() + "\r\n");
    							}
    						}
    					}
    				};
    				sw2.execute();
    			} else if (flag == 1) {
    				JOptionPane.showMessageDialog(null, "不能为空！请重新填写！", "提示消息", JOptionPane.WARNING_MESSAGE);
    			} else if (flag == 2) {
    				JOptionPane.showMessageDialog(null, "相关参数只能是数字！请重新填写！", "提示消息", JOptionPane.WARNING_MESSAGE);
    			} else if (flag == 3) {
    				JOptionPane.showMessageDialog(null, "IP不正确！请重新填写！", "提示消息", JOptionPane.WARNING_MESSAGE);
    			}
    
    		}

这里我们选择直接在方法doInBackground()中处理逻辑以便于前台实时获取实时显示

同理，监听两个面板的中止

// 监听中止
    		if (e.getSource().equals(button5)) {
    			KafkaMainGUI.isRunning2 = false;
    			button4.setEnabled(true);
    			button5.setEnabled(false);
    			button6.setEnabled(true);
    		}

// 监听中止
    		if (e.getSource().equals(button2)) {
    			KafkaMainGUI.isRunning = false;
    			button1.setEnabled(true);
    			button2.setEnabled(false);
    			button3.setEnabled(true);
    			thread.stop();
    		}

监听重置

// 监听重置
    		if (e.getSource().equals(button3)) {
    			tt1.setText("");
    			tt2.setText("");
    			tt3.setText("");
    			tt4.setText("");
    			tt5.setText("");
    			tt6.setText("");
    			tt7.setText("");
    			tt8.setText("");
    			jta1.setText("");
    			jlb11.setText("");
    			jlb11.setVisible(false);
    			jpbProcessLoading1.setVisible(false);
    			jpbProcessLoading1.setValue(0);
    			jrb1.setSelected(true);
    			jta1.setVisible(false);
    			button1.setEnabled(true);
    			button2.setEnabled(false);
    			button3.setEnabled(true);
    		}
    		// 监听重置
    		if (e.getSource().equals(button6)) {
    			tt9.setText("");
    			tt10.setText("");
    			tt11.setText("");
    			tt12.setText("");
    			jta2.setText("");
    			jlb16.setText("");
    			jlb16.setVisible(false);
    			jpbProcessLoading2.setVisible(false);
    			button4.setEnabled(true);
    			button5.setEnabled(false);
    			button6.setEnabled(true);
    		}

监听清除缓存

// 监听清除缓存
    		if (e.getSource().equals(button7)) {
    			jta2.setText("");
    			jlb16.setText("");
    		}

基本监听就是这样了，代码并不全，源码链接GitHub下载：

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

下一章实现后台发送，消费kafka数据。

[https_blog.csdn.net_rico_zhou_article_details_79857625]: https://blog.csdn.net/rico_zhou/article/details/79857625
[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