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

谁践踏了优雅 2022-05-27 00:45 77阅读 0赞

最近公司项目一直需要使用kafka，所以抽空稍微了解了一下kafka，简单介绍下kafka。

简单来说，Kafka是由Apache软件基金会开发的一个开源流处理平台，是一种高吞吐量的分布式发布订阅消息系统，它可以处理

消费者规模的网站中的所有动作流数据。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了

通过集群来提供实时的消息。

本文不着重于kafka，由于公司众多测试小妹妹，有些人没有java环境或者eclipse，测试发送接收kafka数据较为麻烦，所以

做一个简单打包好的exe小程序，方便使用。

首先介绍下kafka的基本参数。

ip：kafka服务器地址

port：kafka默认端口号，默认9092

topic：kafka主题，每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic

key：以json保存，key随意

producer：负责发布消息到Kafka broker

consumer：消息消费者，向Kafka broker读取消息的客户端

consumer GroupId：每个Consumer属于一个特定的Consumer Group

接下来开始撸代码，java小白，请大家多多指教！

IDE使用eclipse，普通javaProject，首先上一张项目结构图：

![20180409113516147][]

简单几个类，分为前台GUI，以及后台主程序处理两个包。

类KafkaMainGUI主要是GUI代码，类Utils主要是一些基本方法，类SendMsgKafka为发送和接收kafka数据的处理代码，类BaseMsg，kafkaMsg，ReturnMsg是实体类，封装一些kafka及前台基本信息。

首先创建基本java项目sendmegtokafka，创建两个包kafka和mainGUI，根目录创建jar包文件夹lib，导入需要的jar包，请到maven中央仓库挨个下载。

![20180409114122256][]

第一个jar包beautyeye\_lnf.jar是javaSwingGUI的第三方jar，更改swing外观，大家根据自己的兴趣可加可不加。

类KafkaMainGUI创建，代码如下：

package mainGUI;
    
    import java.awt.EventQueue;
    import java.awt.Font;
    import java.awt.event.ActionEvent;
    import java.awt.event.ActionListener;
    import java.io.FileNotFoundException;
    import java.io.PrintStream;
    import java.util.Arrays;
    import java.util.Enumeration;
    import java.util.List;
    import java.util.Properties;
    
    import javax.swing.ButtonGroup;
    import javax.swing.ImageIcon;
    import javax.swing.JButton;
    import javax.swing.JFrame;
    import javax.swing.JLabel;
    import javax.swing.JOptionPane;
    import javax.swing.JPanel;
    import javax.swing.JProgressBar;
    import javax.swing.JRadioButton;
    import javax.swing.JScrollPane;
    import javax.swing.JTabbedPane;
    import javax.swing.JTextArea;
    import javax.swing.JTextField;
    import javax.swing.SwingUtilities;
    import javax.swing.SwingWorker;
    import javax.swing.UIManager;
    import javax.swing.plaf.FontUIResource;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.jb2011.lnf.beautyeye.BeautyEyeLNFHelper;
    
    import kafka.BaseMsg;
    import kafka.ReturnMsg;
    import kafka.SendMsgKafka;
    import scala.annotation.StaticAnnotation;
    
    public class KafkaMainGUI extends JFrame implements ActionListener {
    	// 面板
    	public JPanel jp1, jp2;
    	// 标签
    	public JLabel jlb1, jlb2, jlb3, jlb4, jlb5, jlb6, jlb7, jlb8, jlb9, jlb10, jlb11, jlb12, jlb13, jlb14, jlb15, jlb16,
    			jlb17;
    	// 按钮
    	public JButton button1, button2, button3, button4, button5, button6, button7;
    	// 文本域
    	public JTextArea jta1, jta2;
    	// 文本域滚动条
    	public JScrollPane jsp1;
    	// 文本框
    	public JTextField tt1, tt2, tt3, tt4, tt5, tt6, tt7, tt8, tt9, tt10, tt11, tt12, tt13, tt14, tt15, tt16, tt17;
    	// 单选组合
    	public JRadioButton jrb1, jrb2, jrb3;
    	// 单选组合按钮组
    	public ButtonGroup bg1;
    	// 进度条
    	public JProgressBar jpbProcessLoading1, jpbProcessLoading2;
    	// 后台耗时任务线程
    	public SwingWorker<ReturnMsg, String> sw1;
    	public SwingWorker<ReturnMsg, String> sw2;
    	// 实例化kafka主操作类
    	public SendMsgKafka smk = new SendMsgKafka();
    	// 线程状态
    	public static boolean isRunning = true;
    	public static boolean isRunning2 = true;
    	// 组合面板
    	public JTabbedPane jtp = new JTabbedPane(JTabbedPane.TOP);
    	// 实例化工具类
    	public Utils utils = new Utils();
    	// 同步线程
    	public CountThread thread;
    	// 实体消息类
    	public BaseMsg baseMsg, baseMsg2;
    
    	// 构造方法
    	public KafkaMainGUI() {
    		// 初始化
    		init();
    	}
    
    	// 主方法
    	public static void main(String[] args) {
    
    		// 启动gui线程
    		EventQueue.invokeLater(new Runnable() {
    			public void run() {
    				// 设置主题
    				// 初始化字体
    				InitGlobalFont(new Font("微软雅黑", Font.PLAIN, 11));
    				// 设置本属性将改变窗口边框样式定义
    				// 系统默认样式 osLookAndFeelDecorated
    				// 强立体半透明 translucencyAppleLike
    				// 弱立体感半透明 translucencySmallShadow
    				// 普通不透明 generalNoTranslucencyShadow
    				BeautyEyeLNFHelper.frameBorderStyle = BeautyEyeLNFHelper.FrameBorderStyle.generalNoTranslucencyShadow;
    				// 设置主题为BeautyEye
    				try {
    					org.jb2011.lnf.beautyeye.BeautyEyeLNFHelper.launchBeautyEyeLNF();
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    				// 隐藏“设置”按钮
    				UIManager.put("RootPane.setupButtonVisible", false);
    				// 开启/关闭窗口在不活动时的半透明效果
    				// 设置此开关量为false即表示关闭之，BeautyEye LNF中默认是true
    				BeautyEyeLNFHelper.translucencyAtFrameInactive = false;
    				// 设置BeantuEye外观下JTabbedPane的左缩进
    				// 改变InsetsUIResource参数的值即可实现
    				UIManager.put("TabbedPane.tabAreaInsets", new javax.swing.plaf.InsetsUIResource(3, 20, 2, 20));
    				// UIManager.setLookAndFeel(UIManager.getSystemLookAndFeelClassName());
    
    				// 启动主界面
    				new KafkaMainGUI();
    			}
    		});
    	}
    
    	// font
    	private static void InitGlobalFont(Font font) {
    		FontUIResource fontRes = new FontUIResource(font);
    		for (Enumeration<Object> keys = UIManager.getDefaults().keys(); keys.hasMoreElements();) {
    			Object key = keys.nextElement();
    			Object value = UIManager.get(key);
    			if (value instanceof FontUIResource) {
    				UIManager.put(key, fontRes);
    			}
    		}
    	}
    
    	// 初始化
    	private void init() {
    		// 两个面板，一个用于发送数据，一个用于接收数据
    		jp1 = new JPanel();
    		jp2 = new JPanel();
    		// 布局采用最简单的绝对布局
    		jp1.setLayout(null);
    		jp2.setLayout(null);
    
    		// 主界面设计
    		jlb1 = new JLabel("IP：");
    		// 位置
    		jlb1.setBounds(20, 20, 100, 25);
    		// 添加到面板
    		jp1.add(jlb1);
    		tt1 = new JTextField();
    		tt1.setBounds(80, 20, 150, 25);
    		// 默认项
    		tt1.setText("192.168.1.140");
    		jp1.add(tt1);
    
    		jlb2 = new JLabel("PORT：");
    		jlb2.setBounds(250, 20, 100, 25);
    		jp1.add(jlb2);
    		tt2 = new JTextField();
    		tt2.setBounds(320, 20, 150, 25);
    		tt2.setText("9092");
    		jp1.add(tt2);
    
    		jlb3 = new JLabel("GroupId：");
    		jlb3.setBounds(20, 60, 100, 25);
    		jp1.add(jlb3);
    		tt3 = new JTextField();
    		tt3.setBounds(80, 60, 150, 25);
    		tt3.setText("groupid");
    		jp1.add(tt3);
    
    		jlb4 = new JLabel("Topic：");
    		jlb4.setBounds(250, 60, 100, 25);
    		jp1.add(jlb4);
    		tt4 = new JTextField();
    		tt4.setBounds(320, 60, 150, 25);
    		tt4.setText("test1");
    		jp1.add(tt4);
    
    		jlb5 = new JLabel("条数：");
    		jlb5.setBounds(20, 100, 100, 25);
    		jp1.add(jlb5);
    		tt5 = new JTextField();
    		tt5.setBounds(80, 100, 150, 25);
    		tt5.setText("10");
    		jp1.add(tt5);
    
    		jlb6 = new JLabel("间隔时间：");
    		jlb6.setBounds(250, 100, 100, 25);
    		jp1.add(jlb6);
    		tt6 = new JTextField();
    		tt6.setBounds(320, 100, 120, 25);
    		tt6.setText("1000");
    		jp1.add(tt6);
    
    		jlb7 = new JLabel("(ms)");
    		jlb7.setBounds(440, 100, 35, 25);
    		jp1.add(jlb7);
    
    		jta1 = new JTextArea();
    		jta1.setBounds(80, 180, 360, 250);
    		jta1.setFont(new Font("宋体", Font.PLAIN, 20));
    		jp1.add(jta1);
    		jta1.setVisible(false);
    
    		jlb8 = new JLabel("Msg：");
    		jlb8.setBounds(20, 140, 100, 25);
    		jp1.add(jlb8);
    		jrb1 = new JRadioButton("默认");
    		jrb1.setBounds(80, 140, 60, 25);
    		// 监听单选框，选中第一个则进行什么操作，选中第二个又进行什么操作
    		jrb1.addActionListener(new ActionListener() {
    			@Override
    			public void actionPerformed(ActionEvent e) {
    				if (jrb1.isSelected()) {
    					jta1.setVisible(false);
    				} else {
    					jta1.setVisible(true);
    				}
    			}
    		});
    		jp1.add(jrb1);
    		jrb2 = new JRadioButton("自定义");
    		jrb2.setBounds(170, 140, 80, 25);
    		jrb2.addActionListener(new ActionListener() {
    			@Override
    			public void actionPerformed(ActionEvent e) {
    				if (jrb2.isSelected()) {
    					jta1.setVisible(true);
    				} else {
    					jta1.setVisible(false);
    				}
    			}
    		});
    		jp1.add(jrb2);
    		// 加入按钮组，只能选中一个
    		bg1 = new ButtonGroup();
    		bg1.add(jrb1);
    		bg1.add(jrb2);
    		jrb1.setSelected(true);
    
    		jlb9 = new JLabel("Key：");
    		jlb9.setBounds(250, 140, 100, 25);
    		jp1.add(jlb9);
    		tt7 = new JTextField();
    		tt7.setBounds(295, 140, 60, 25);
    		tt7.setText("key");
    		jp1.add(tt7);
    
    		jlb10 = new JLabel("StartId：");
    		jlb10.setBounds(360, 140, 100, 25);
    		jp1.add(jlb10);
    		tt8 = new JTextField();
    		tt8.setBounds(410, 140, 60, 25);
    		tt8.setText("1");
    		jp1.add(tt8);
    
    		// 设置滚动条
    		jpbProcessLoading1 = new JProgressBar();
    		// 可呈现字符串
    		jpbProcessLoading1.setStringPainted(true);
    		jpbProcessLoading1.setBounds(100, 440, 300, 25);
    		// 设置最大最小value
    		jpbProcessLoading1.setMinimum(1);
    		jpbProcessLoading1.setMaximum(100);
    		// 主界面初始化时不可见
    		jpbProcessLoading1.setVisible(false);
    		jp1.add(jpbProcessLoading1);
    
    		jlb11 = new JLabel();
    		jlb11.setBounds(410, 440, 100, 25);
    		jp1.add(jlb11);
    		jlb11.setVisible(false);
    
    		button1 = new JButton("发送");
    		button1.setBounds(50, 480, 80, 25);
    		// 启动监听按钮
    		button1.addActionListener(this);
    		jp1.add(button1);
    		button1.setEnabled(true);
    
    		button2 = new JButton("中止");
    		button2.setBounds(200, 480, 80, 25);
    		button2.addActionListener(this);
    		jp1.add(button2);
    		button2.setVisible(true);
    		button2.setEnabled(false);
    
    		button3 = new JButton("重置");
    		button3.setBounds(350, 480, 80, 25);
    		button3.addActionListener(this);
    		jp1.add(button3);
    		button3.setVisible(true);
    		button3.setEnabled(true);
    
    		// 设置接收消息的面板jp2
    		jlb12 = new JLabel("IP：");
    		jlb12.setBounds(20, 20, 100, 25);
    		jp2.add(jlb12);
    		tt9 = new JTextField();
    		tt9.setBounds(80, 20, 150, 25);
    		tt9.setText("192.168.1.140");
    		jp2.add(tt9);
    
    		jlb13 = new JLabel("PORT：");
    		jlb13.setBounds(250, 20, 100, 25);
    		jp2.add(jlb13);
    		tt10 = new JTextField();
    		tt10.setBounds(320, 20, 150, 25);
    		tt10.setText("9092");
    		jp2.add(tt10);
    
    		jlb14 = new JLabel("GroupId：");
    		jlb14.setBounds(20, 60, 100, 25);
    		jp2.add(jlb14);
    		tt11 = new JTextField();
    		tt11.setBounds(80, 60, 150, 25);
    		tt11.setText("groupid2");
    		jp2.add(tt11);
    
    		jlb15 = new JLabel("Topic：");
    		jlb15.setBounds(250, 60, 100, 25);
    		jp2.add(jlb15);
    		tt12 = new JTextField();
    		tt12.setBounds(320, 60, 150, 25);
    		tt12.setText("test1");
    		jp2.add(tt12);
    
    		// 消息区
    		jta2 = new JTextArea();
    		// 设置为不可编辑
    		jta2.setEditable(false);
    		// 将文本域加入滚动条，因为接受的消息较多会超过文本域范围
    		jsp1 = new JScrollPane(jta2);
    		jsp1.setBounds(20, 100, 450, 300);
    		jta1.setFont(new Font("宋体", Font.PLAIN, 15));
    		jp2.add(jsp1);
    
    		button7 = new JButton("清除缓存");
    		button7.setBounds(50, 440, 100, 25);
    		button7.addActionListener(this);
    		jp2.add(button7);
    		button7.setEnabled(true);
    
    		jlb16 = new JLabel();
    		jlb16.setBounds(410, 440, 100, 25);
    		jp2.add(jlb16);
    		jlb16.setVisible(false);
    
    		jpbProcessLoading2 = new JProgressBar();
    		jpbProcessLoading2.setStringPainted(true);
    		jpbProcessLoading2.setBounds(160, 440, 200, 25);
    		jpbProcessLoading2.setVisible(false);
    		jp2.add(jpbProcessLoading2);
    
    		button4 = new JButton("消费");
    		button4.setBounds(50, 480, 80, 25);
    		button4.addActionListener(this);
    		jp2.add(button4);
    		button4.setEnabled(true);
    
    		button5 = new JButton("中止");
    		button5.setBounds(200, 480, 80, 25);
    		button5.addActionListener(this);
    		jp2.add(button5);
    		button5.setVisible(true);
    		button5.setEnabled(false);
    
    		button6 = new JButton("重置");
    		button6.setBounds(350, 480, 80, 25);
    		button6.addActionListener(this);
    		jp2.add(button6);
    		button6.setVisible(true);
    		button6.setEnabled(true);
    
    		// this表示jframe，将组合面板加入窗口
    		this.add(jtp);
    		// 设置面板的基本显示信息
    		jtp.add("Send Message To Kafka", jp1);
    		jtp.add("Get Message From Kafka", jp2);
    		// 窗口标题
    		this.setTitle("Kafka");
    		// 窗口大小
    		this.setSize(500, 600);
    		// 居中显示
    		this.setLocationRelativeTo(null);
    		this.setVisible(true);
    		// 大小不可改变
    		this.setResizable(false);
    		// 点击x则退出程序
    		this.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
    		// 设置取代java图标
    		ImageIcon i = new ImageIcon(getClass().getResource("kafka.png"));
    		this.setIconImage(i.getImage());
    	}
    
    	@Override
    	public void actionPerformed(ActionEvent e) {
    		
    	}
    	
    }

基本界面就是这样，主要的逻辑操作在监听方法void actionPerformed(ActionEvent e)中书写，稍后再处理。

现在来创建实体类BaseMsg，用于封装前台界面的参数

package kafka;
    
    public class BaseMsg {
    	// ip
    	public String kafkaIp;
    	// port
    	public String kafkaPort;
    	// groupid
    	public String kafkaGroupId;
    	// topic
    	public String kafkaTopic;
    	// account
    	public String kafkaAccount;
    	// times
    	public String kafkaTimes;
    	// is msg type
    	public boolean kafkaIsDefault;
    	// msg
    	public String kafkaMessage;
    	// key
    	public String kafkaKey;
    	// startid
    	public String kafkaStartId;
    
    	public String getKafkaStartId() {
    		return kafkaStartId;
    	}
    
    	public void setKafkaStartId(String kafkaStartId) {
    		this.kafkaStartId = kafkaStartId;
    	}
    
    	public String getKafkaIp() {
    		return kafkaIp;
    	}
    
    	public void setKafkaIp(String kafkaIp) {
    		this.kafkaIp = kafkaIp;
    	}
    
    	public String getKafkaPort() {
    		return kafkaPort;
    	}
    
    	public void setKafkaPort(String kafkaPort) {
    		this.kafkaPort = kafkaPort;
    	}
    
    	public String getKafkaGroupId() {
    		return kafkaGroupId;
    	}
    
    	public void setKafkaGroupId(String kafkaGroupId) {
    		this.kafkaGroupId = kafkaGroupId;
    	}
    
    	public String getKafkaTopic() {
    		return kafkaTopic;
    	}
    
    	public void setKafkaTopic(String kafkaTopic) {
    		this.kafkaTopic = kafkaTopic;
    	}
    
    	public String getKafkaAccount() {
    		return kafkaAccount;
    	}
    
    	public void setKafkaAccount(String kafkaAccount) {
    		this.kafkaAccount = kafkaAccount;
    	}
    
    	public String getKafkaTimes() {
    		return kafkaTimes;
    	}
    
    	public void setKafkaTimes(String kafkaTimes) {
    		this.kafkaTimes = kafkaTimes;
    	}
    
    	public boolean isKafkaIsDefault() {
    		return kafkaIsDefault;
    	}
    
    	public void setKafkaIsDefault(boolean kafkaIsDefault) {
    		this.kafkaIsDefault = kafkaIsDefault;
    	}
    
    	public String getKafkaMessage() {
    		return kafkaMessage;
    	}
    
    	public void setKafkaMessage(String kafkaMessage) {
    		this.kafkaMessage = kafkaMessage;
    	}
    
    	public String getKafkaKey() {
    		return kafkaKey;
    	}
    
    	public void setKafkaKey(String kafkaKey) {
    		this.kafkaKey = kafkaKey;
    	}
    
    	@Override
    	public String toString() {
    		return "BaseMsg [kafkaIp=" + kafkaIp + ", kafkaPort=" + kafkaPort + ", kafkaGroupId=" + kafkaGroupId
    				+ ", kafkaTopic=" + kafkaTopic + ", kafkaAccount=" + kafkaAccount + ", kafkaTimes=" + kafkaTimes
    				+ ", kafkaIsDefault=" + kafkaIsDefault + ", kafkaMessage=" + kafkaMessage + ", kafkaKey=" + kafkaKey
    				+ ", kafkaStartId=" + kafkaStartId + "]";
    	}
    
    }

后台处理好前台的参数实体类kafkaMsg

package kafka;
    
    public class kafkaMsg {
    	// ip:port
    	public String ipPort;
    	// groupid
    	public String groupId;
    	// topic
    	public String topic;
    	// key
    	public String key;
    	// account
    	public int account;
    	// time
    	public int time;
    	// msg
    	public String message;
    	// startid
    	public int startId;
    
    	public String getIpPort() {
    		return ipPort;
    	}
    
    	public void setIpPort(String ipPort) {
    		this.ipPort = ipPort;
    	}
    
    	public String getGroupId() {
    		return groupId;
    	}
    
    	public void setGroupId(String groupId) {
    		this.groupId = groupId;
    	}
    
    	public String getTopic() {
    		return topic;
    	}
    
    	public void setTopic(String topic) {
    		this.topic = topic;
    	}
    
    	public String getKey() {
    		return key;
    	}
    
    	public void setKey(String key) {
    		this.key = key;
    	}
    
    	public int getAccount() {
    		return account;
    	}
    
    	public void setAccount(int account) {
    		this.account = account;
    	}
    
    	public int getTime() {
    		return time;
    	}
    
    	public void setTime(int time) {
    		this.time = time;
    	}
    
    	public String getMessage() {
    		return message;
    	}
    
    	public void setMessage(String message) {
    		this.message = message;
    	}
    
    	public int getStartId() {
    		return startId;
    	}
    
    	public void setStartId(int startId) {
    		this.startId = startId;
    	}
    
    	@Override
    	public String toString() {
    		return "kafkaMsg [ipPort=" + ipPort + ", groupId=" + groupId + ", topic=" + topic + ", key=" + key
    				+ ", account=" + account + ", time=" + time + ", message=" + message + ", startId=" + startId + "]";
    	}
    
    }

返回数据封装类ReturnMsg

package kafka;
    
    import javax.imageio.event.IIOReadProgressListener;
    
    public class ReturnMsg {
    	//是否完成
    	public boolean isCompleted;
    	//完成的条数
    	public int comTimes;
    
    	public boolean isCompleted() {
    		return isCompleted;
    	}
    
    	public void setCompleted(boolean isCompleted) {
    		this.isCompleted = isCompleted;
    	}
    
    	public int getComTimes() {
    		return comTimes;
    	}
    
    	public void setComTimes(int comTimes) {
    		this.comTimes = comTimes;
    	}
    
    	@Override
    	public String toString() {
    		return "ReturnMsg [isCompleted=" + isCompleted + ", comTimes=" + comTimes + "]";
    	}
    
    }

基本界面完成，再创建一个工具类Utils，运行KafkaMainGUI.class即可出现以下界面。

![20180409131617890][]

![20180409131703897][]

接下来进行逻辑处理，实现发送功能。

源码地址GitHub：[https://github.com/ricozhou/msgkafka][https_github.com_ricozhou_msgkafka]

待续。

[20180409113516147]: /images/20220527/665c05c646b14186b8fafd88a0eeb546.png
[20180409114122256]: /images/20220527/4787613d198a47db80c8a7bcf0d053c7.png
[20180409131617890]: /images/20220527/b675187ec0e6422a93f3aaaa0db05b3d.png
[20180409131703897]: /images/20220527/f70a008887a043c7ab4f645b21817094.png
[https_github.com_ricozhou_msgkafka]: https://github.com/ricozhou/msgkafka