关键词搜索

源码搜索 ×
×

JAVA通信编程（四）——UDP通讯

发布2015-12-05浏览4333次

详情内容

欢迎支持笔者新作：《深入理解Kafka:核心设计与实践原理》和《RabbitMQ实战指南》，同时欢迎关注笔者的微信公众号：朱小厮的博客。

欢迎跳转到本文的原文链接：https://honeypps.com/network/java-udp-comm/

 

经过TCP和串口通讯编程的了解，相信大家应该掌握CommBuff的套路了，这里首先展示的是通过UDP编程的方式实现CommBuff接口，之后通过简单工厂模式的应用说明如何屏蔽底层通讯差异。

UdpImpl类如下：

 

  1. package com.zzh.comm;
  2.  
  3. import java.io.IOException;
  4. import java.net.DatagramPacket;
  5. import java.net.DatagramSocket;
  6. import java.net.InetAddress;
  7. import java.net.SocketException;
  8. import java.net.UnknownHostException;
  9. import java.util.Map;
  10.  
  11. import org.apache.log4j.Logger;
  12.  
  13. public class UdpImpl implements CommBuff
  14. {
  15. 	private Logger logger = Logger.getLogger(Object.class.getName());
  16. 	
  17. 	private int local_port;
  18. 	private int dest_port;
  19. 	private String ip;
  20. 	private int time_out;
  21. 	
  22. 	DatagramSocket client = null;
  23. 	
  24. 	private String fileName = "/udp.properties";
  25. 	public UdpImpl()
  26. 	{
  27. 		Map<String,String> map = new ReadProperties().getPropertiesMap(fileName);
  28. 		try
  29. 		{
  30. 			local_port = Integer.parseInt(map.get("udp_local_port"));
  31. 			dest_port = Integer.parseInt(map.get("udp_dest_port"));
  32. 			time_out = Integer.parseInt(map.get("udp_timeout"));
  33. 			ip = map.get("udp_dest_ip");
  34. 		}
  35. 		catch (Exception e)
  36. 		{
  37. 			logger.error(e.getMessage());
  38. 		}
  39. 	}
  40. 	
  41. 	@Override
  42. 	public byte[] readBuff()
  43. 	{
  44. 		if(client == null)
  45. 		{
  46. 			throw new RuntimeException("clinet is null!");
  47. 		}
  48. 		byte[] recvBuf = new byte[1024];
  49. 		DatagramPacket recvPacket = new DatagramPacket(recvBuf , recvBuf.length);
  50. 		try
  51. 		{
  52. 			client.receive(recvPacket);
  53. 		}
  54. 		catch (IOException e)
  55. 		{
  56. 			logger.info(e.getMessage());
  57. 			return new byte[0];
  58. 		}
  59. 		byte[] ans = new byte[recvPacket.getLength()];
  60. 		System.arraycopy(recvPacket.getData(), 0, ans, 0, recvPacket.getLength());
  61. 		logger.info("网口接收："+CommUtil.bytesToHex(ans));
  62. 		return ans;
  63. 	}
  64.  
  65. 	@Override
  66. 	public void writeBuff(byte[] message)
  67. 	{
  68. 		if(client == null)
  69. 		{
  70. 			throw new RuntimeException("clinet is null!");
  71. 		}
  72. 		
  73. 		try
  74. 		{
  75. 			InetAddress addr = InetAddress.getByName(ip);
  76. 			DatagramPacket sendPacket = new DatagramPacket(message,message.length,addr,dest_port);
  77. 			client.send(sendPacket);
  78. 			logger.info("发送成功: "+CommUtil.bytesToHex(message));
  79. 		}
  80. 		catch (UnknownHostException e)
  81. 		{
  82. 			logger.error(e.getMessage());
  83. 		}
  84. 		catch (IOException e)
  85. 		{
  86. 			logger.error(e.getMessage());
  87. 		}
  88. 		
  89. 	}
  90.  
  91. 	@Override
  92. 	public void open() {
  93. 		try
  94. 		{
  95. 			client = new DatagramSocket(local_port);
  96. 			client.setSoTimeout(time_out);
  97. 			if(client != null)
  98. 			{
  99. 				logger.info("client open succeed!");
  100. 			}
  101. 		}
  102. 		catch (SocketException e)
  103. 		{
  104. 			logger.error(e.getMessage());
  105. 		}
  106. 	}
  107.  
  108. 	@Override
  109. 	public void close() 
  110. 	{
  111. 		if(client != null)
  112. 		{
  113. 			client.close();
  114. 		}
  115. 	}
  116.  
  117. 	@Override
  118. 	public Object getInfo()
  119. 	{
  120. 		return null;
  121. 	}
  122.  
  123. }

UdpImpl实现了CommBuff接口的各个方法。UDP Socket采用的数据包的方式进行通讯的，这个可以与TCP的方式区分开。

 

下面通过一个简单工厂模式，可以实现底层通讯的便利性。

 

  1. package com.zzh.comm;
  2.  
  3. public class CommFactory
  4. {
  5. 	public CommBuff getCommBuff(String properties) throws Exception
  6. 	{
  7. 		if(properties.equals("comm_serial"))
  8. 		{
  9. 			return new SerialImpl();
  10. 		}
  11. 		else if(properties.equals("comm_tcpServer"))
  12. 		{
  13. 			return new TcpServerImpl();
  14. 		}
  15. 		else if(properties.equals("comm_tcpClient"))
  16. 		{
  17. 			return new TcpClientImpl();
  18. 		}
  19. 		else if(properties.equals("comm_udp"))
  20. 		{
  21. 			return new UdpImpl();
  22. 		}
  23. 		else
  24. 		{
  25. 			throw new Exception("Communication para error: no found avaliable communication Object instance.");
  26. 		}
  27. 	}
  28. }

上面的getCommBuff方法通过参数properties可以初始化不同的通讯接口实现类，这样上次应用只需调用Commbuff接口的方法，而无需与底层通讯的细节相融合，极大的降低了程序间的耦合性。

 

本篇就简单的阐述到这里。但是下面会附加一个程序，这个程序通过调用CommFactory的方法生成底层通讯的实例，程序的主要内容是电力行业的某个通讯规约（Modbus）的实现，如果非电力行业的通讯，可以不必了解程序中的细节，可以大概看一下怎么使用.

 

  1. package com.zzh.protocol;
  2.  
  3. import java.util.Calendar;
  4. import java.util.Map;
  5. import java.util.concurrent.ConcurrentLinkedDeque;
  6. import java.util.concurrent.ExecutorService;
  7. import java.util.concurrent.Executors;
  8. import java.util.concurrent.TimeUnit;
  9.  
  10. import com.zzh.comm.CommBuff;
  11. import com.zzh.comm.CommFactory;
  12. import com.zzh.comm.CommUtil;
  13. import com.zzh.comm.ReadProperties;
  14. import com.zzh.dao.ModbusDao;
  15. import com.zzh.dao.ModbusDaoImpl;
  16. import com.zzh.dao.pojo.ModbusPojo;
  17.  
  18. public class Modbus {
  19. 	private CommBuff comm;
  20. 	private int comm_timeout;
  21. 	private byte devAddr;
  22. 	
  23. 	private static int RECV_SIZE = 35;
  24. 	private static int RECV_INNER_SIZE = 30;
  25. 	private static int MINUTE=60000;
  26. 	private volatile boolean  refreshFlag = false;
  27. 	
  28. 	private ModbusPojo modbusPojo; 
  29. 	
  30. 	private ConcurrentLinkedDeque<Byte> deque = new ConcurrentLinkedDeque<Byte>();
  31. 	private String fileName = "/modbus.properties";
  32. 	
  33. 	public Modbus()
  34. 	{
  35. 		Map<String,String> map = new ReadProperties().getPropertiesMap(fileName);
  36. 		String comm_way = map.get("modbus_comm_way");
  37. 		String comm_timeouts = map.get("comm_timeout");
  38. 		comm_timeout = Integer.parseInt(comm_timeouts);
  39. 		String devAddrs = map.get("devAddr");
  40. 		devAddr = Byte.parseByte(devAddrs);
  41. 		if(comm_way!=null)
  42. 		{
  43. 			modbusPojo = new ModbusPojo(); 
  44. 			try
  45. 			{
  46. 				comm = new CommFactory().getCommBuff(comm_way);
  47. 			}
  48. 			catch (Exception e)
  49. 			{
  50. 				e.printStackTrace();
  51. 			}
  52. 			comm.open();
  53. 			
  54. 			ExecutorService pool = Executors.newFixedThreadPool(2);
  55. 			Thread thread1 = new Thread(new readThread());
  56. 	    	thread1.setDaemon(true);
  57. 	    	Thread thread2 = new Thread(new dbThread());
  58. 	    	thread2.setDaemon(true);
  59. 	    	pool.execute(thread1);
  60. 	    	pool.execute(thread2);
  61. 		}
  62. 		else
  63. 		{
  64. 			throw new RuntimeException("没有配置好合适的串口参数");
  65. 		}
  66. 	}
  67. 	
  68. 	private class readThread implements Runnable
  69. 	{
  70. 		@Override
  71. 		public void run()
  72. 		{
  73. 			while(true)
  74. 			{
  75. 				byte[] recvBuff = comm.readBuff();
  76. 				if(recvBuff.length>0)
  77. 				{
  78. 					for(int i=0;i<recvBuff.length;i++)
  79. 					{
  80. 						deque.add(recvBuff[i]);
  81. 					}
  82. 				}
  83. 				try
  84. 				{
  85. 					TimeUnit.MILLISECONDS.sleep(1000);
  86. 				}
  87. 				catch (InterruptedException e)
  88. 				{
  89. 					e.printStackTrace();
  90. 				}
  91. 			}
  92. 		}
  93. 	}
  94. 	
  95. 	private class dbThread implements Runnable
  96. 	{
  97. 		@Override
  98. 		public void run()
  99. 		{
  100. 			while(true)
  101. 			{
  102. 				if(refreshFlag == true)
  103. 				{
  104. 					Calendar now = Calendar.getInstance();
  105. 					if(now.get(Calendar.MINUTE)%5==0)
  106. //					if(true)
  107. 					{
  108. 						synchronized (modbusPojo)
  109. 						{
  110. 							filterModbusPojo();
  111. 							modbusPojo.setNow(TimeUtil.getDateOfMM(now));
  112. //							modbusPojo.setNow(new java.sql.Timestamp(new Date().getTime()));
  113. 							ModbusDao md = new ModbusDaoImpl();
  114. 							md.addModbus(modbusPojo);
  115. 						}
  116. 					}
  117. 				}
  118. 				try
  119. 				{
  120. 					TimeUnit.MILLISECONDS.sleep(MINUTE);
  121. //					TimeUnit.MILLISECONDS.sleep(1000);
  122. 				}
  123. 				catch (InterruptedException e)
  124. 				{
  125. 					e.printStackTrace();
  126. 				}
  127. 			}
  128. 		}
  129. 		
  130. 	}
  131. 	
  132. 	public void filterModbusPojo()
  133. 	{
  134. 		modbusPojo.setQua(0);
  135. 		if(modbusPojo.getEnvTemperature()>ModbusUtil.TEMPERATURE_UP)
  136. 		{
  137. 			modbusPojo.setEnvTemperature(ModbusUtil.TEMPERATURE_UP);
  138. 			System.out.println("getEnvTemperature = "+modbusPojo.getEnvTemperature());
  139. 			modbusPojo.setQua(1);
  140. 		}
  141. 		if(modbusPojo.getEnvTemperature()<ModbusUtil.TEMPERATURE_LOW)
  142. 		{
  143. 			modbusPojo.setEnvTemperature(ModbusUtil.TEMPERATURE_LOW);
  144. 			System.out.println("getEnvTemperature = "+modbusPojo.getEnvTemperature());
  145. 			modbusPojo.setQua(1);
  146. 		}
  147. 		if(modbusPojo.getTemperature()>ModbusUtil.TEMPERATURE_UP)
  148. 		{
  149. 			modbusPojo.setTemperature(ModbusUtil.TEMPERATURE_UP);
  150. 			System.out.println("getTemperature = "+modbusPojo.getTemperature());
  151. 			modbusPojo.setQua(1);
  152. 		}
  153. 		if(modbusPojo.getTemperature()<ModbusUtil.TEMPERATURE_LOW)
  154. 		{
  155. 			modbusPojo.setTemperature(ModbusUtil.TEMPERATURE_LOW);
  156. 			System.out.println("getTemperature = "+modbusPojo.getTemperature());
  157. 			modbusPojo.setQua(1);
  158. 		}
  159. 		if(modbusPojo.getHumidity()>ModbusUtil.HUMIDITY_UP)
  160. 		{
  161. 			modbusPojo.setHumidity(ModbusUtil.HUMIDITY_UP);
  162. 			System.out.println("getHumidity = "+modbusPojo.getHumidity());
  163. 			modbusPojo.setQua(1);
  164. 		}
  165. 		if(modbusPojo.getHumidity()<ModbusUtil.HUMIDITY_LOW)
  166. 		{
  167. 			modbusPojo.setHumidity(ModbusUtil.HUMIDITY_LOW);
  168. 			System.out.println("getHumidity = "+modbusPojo.getHumidity());
  169. 			modbusPojo.setQua(1);
  170. 		}
  171. 		if(modbusPojo.getPressure()>ModbusUtil.PRESSURE_UP)
  172. 		{
  173. 			modbusPojo.setPressure(ModbusUtil.PRESSURE_UP);
  174. 			System.out.println("getPressure = "+modbusPojo.getPressure());
  175. 			modbusPojo.setQua(1);
  176. 		}
  177. 		if(modbusPojo.getPressure()<ModbusUtil.PRESSURE_LOW)
  178. 		{
  179. 			modbusPojo.setPressure(ModbusUtil.PRESSURE_LOW);
  180. 			System.out.println("getPressure = "+modbusPojo.getPressure());
  181. 			modbusPojo.setQua(1);
  182. 		}
  183. 		if(modbusPojo.getIrradiance()>ModbusUtil.IRRADIANCE_UP)
  184. 		{
  185. 			modbusPojo.setIrradiance(ModbusUtil.IRRADIANCE_UP);
  186. 			System.out.println("getIrradiance = "+modbusPojo.getIrradiance());
  187. 			modbusPojo.setQua(1);
  188. 		}
  189. 		if(modbusPojo.getIrradiance()<ModbusUtil.IRRADIANCE_LOW)
  190. 		{
  191. 			modbusPojo.setIrradiance(ModbusUtil.IRRADIANCE_LOW);
  192. 			System.out.println("getIrradiance = "+modbusPojo.getIrradiance());
  193. 			modbusPojo.setQua(1);
  194. 		}
  195. 		if(modbusPojo.getScaIrradiance()>ModbusUtil.IRRADIANCE_UP)
  196. 		{
  197. 			modbusPojo.setScaIrradiance(ModbusUtil.IRRADIANCE_UP);
  198. 			System.out.println("getScaIrradiance = "+modbusPojo.getScaIrradiance());
  199. 			modbusPojo.setQua(1);
  200. 		}
  201. 		if(modbusPojo.getScaIrradiance()<ModbusUtil.IRRADIANCE_LOW)
  202. 		{
  203. 			modbusPojo.setScaIrradiance(ModbusUtil.IRRADIANCE_LOW);
  204. 			System.out.println("getScaIrradiance = "+modbusPojo.getScaIrradiance());
  205. 			modbusPojo.setQua(1);
  206. 		}
  207. 		if(modbusPojo.getDirIrradiance()>ModbusUtil.IRRADIANCE_UP)
  208. 		{
  209. 			modbusPojo.setDirIrradiance(ModbusUtil.IRRADIANCE_UP);
  210. 			System.out.println("getDirIrradiance = "+modbusPojo.getDirIrradiance());
  211. 			modbusPojo.setQua(1);
  212. 		}
  213. 		if(modbusPojo.getDirIrradiance()<ModbusUtil.IRRADIANCE_LOW)
  214. 		{
  215. 			modbusPojo.setDirIrradiance(ModbusUtil.IRRADIANCE_LOW);
  216. 			System.out.println("getDirIrradiance = "+modbusPojo.getDirIrradiance());
  217. 			modbusPojo.setQua(1);
  218. 		}
  219. 		if(modbusPojo.getWindSpeed()>ModbusUtil.UAVG_UP)
  220. 		{
  221. 			modbusPojo.setWindSpeed(ModbusUtil.UAVG_UP);
  222. 			System.out.println("getWindSpeed = "+modbusPojo.getWindSpeed());
  223. 			modbusPojo.setQua(1);
  224. 		}
  225. 		if(modbusPojo.getWindSpeed()<ModbusUtil.UAVG_LOW)
  226. 		{
  227. 			modbusPojo.setWindSpeed(ModbusUtil.UAVG_LOW);
  228. 			System.out.println("getWindSpeed = "+modbusPojo.getWindSpeed());
  229. 			modbusPojo.setQua(1);
  230. 		}
  231. 		if(modbusPojo.getWindDir()>ModbusUtil.VAVG_UP)
  232. 		{
  233. 			modbusPojo.setWindDir(ModbusUtil.VAVG_UP);
  234. 			System.out.println("getWindDir = "+modbusPojo.getWindDir());
  235. 			modbusPojo.setQua(1);
  236. 		}
  237. 		if(modbusPojo.getWindDir()<ModbusUtil.VAVG_LOW)
  238. 		{
  239. 			modbusPojo.setWindDir(ModbusUtil.VAVG_LOW);
  240. 			System.out.println("getWindDir = "+modbusPojo.getWindDir());
  241. 			modbusPojo.setQua(1);
  242. 		}
  243. 	}
  244. 	
  245. 	public void process()
  246. 	{
  247. 		try
  248. 		{
  249. 			TimeUnit.MILLISECONDS.sleep(comm_timeout);
  250. 		}
  251. 		catch (InterruptedException e)
  252. 		{
  253. 			e.printStackTrace();
  254. 		}
  255. 		recvProcess();
  256. 		sendProcess();
  257. 	}
  258.  
  259. 	public void recvProcess()
  260. 	{
  261. 		refreshFlag = false;
  262. 		byte[] recvBuff = new byte[RECV_INNER_SIZE];
  263. 		while(deque.size()>=RECV_SIZE)
  264. 		{
  265. 			Byte first = deque.pollFirst();
  266. 			if(first == devAddr)
  267. 			{
  268. 				Byte second = deque.pollFirst();
  269. 				if(second == 0x03)
  270. 				{
  271. 					Byte third = deque.pollFirst();
  272. 					if(third == RECV_INNER_SIZE)
  273. 					{
  274. 						for(int i=0;i<RECV_INNER_SIZE;i++)
  275. 						{
  276. 							recvBuff[i] = deque.pollFirst();
  277. 						}
  278. 						deque.pollFirst();
  279. 						deque.pollFirst();
  280. 						dealRecvBuff(recvBuff);
  281. 					}
  282. 				}
  283. 			}
  284. 		}
  285. 	}
  286. 	
  287. 	public void dealRecvBuff(byte[] recvBuff)
  288. 	{
  289. 		System.out.println(CommUtil.bytesToHex(recvBuff));
  290. 		refreshFlag = true;
  291. 		getModbusPojo(recvBuff);
  292. //		modbusPojo.print();
  293. 	}
  294. 	
  295. 	public void getModbusPojo(byte[] recvBuff)
  296. 	{
  297. 		int temp;
  298. 		synchronized (modbusPojo)
  299. 		{
  300. 			for(int i=0;i<recvBuff.length;)
  301. 			{
  302. 				switch(i)
  303. 				{
  304. 					case 0:
  305. 						temp = ModbusUtil.getSignedAns(recvBuff, 0, 1);
  306. 						double envTemperature = temp*0.1;
  307. 						modbusPojo.setEnvTemperature(envTemperature);
  308. 						break;
  309. 					case 2:
  310. 						temp = ModbusUtil.getSignedAns(recvBuff, 2, 3);
  311. 						double temperature = temp*0.1;
  312. 						modbusPojo.setTemperature(temperature);
  313. 						break;
  314. 					case 4:
  315. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 4, 5);
  316. 						double humidity = temp*0.1;
  317. 						modbusPojo.setHumidity(humidity);
  318. 						break;
  319. 					case 6:
  320. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 6, 7);
  321. 						double pressure = temp*0.1;
  322. 						modbusPojo.setPressure(pressure);
  323. 						break;
  324. 					case 8:
  325. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 8, 9);
  326. 						modbusPojo.setIrradiance(temp);
  327. 						break;
  328. 					case 10:
  329. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 10, 11);
  330. 						modbusPojo.setScaIrradiance(temp);
  331. 						break;
  332. 					case 12:
  333. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 12, 13);
  334. 						modbusPojo.setDirIrradiance(temp);
  335. 						break;
  336. 					case 14:
  337. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 14, 15);
  338. 						modbusPojo.setWindDir(temp);
  339. 						break;
  340. 					case 16:
  341. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 16, 17);
  342. 						double windSpeed = temp*0.1;
  343. 						modbusPojo.setWindSpeed(windSpeed);
  344. 						break;
  345. 					case 18:
  346. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 18, 19);
  347. 						double windSpeedTwo = temp*0.1;
  348. 						modbusPojo.setWindSpeedTwo(windSpeedTwo);
  349. 						break;
  350. 					case 20:
  351. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 20, 21);
  352. 						double windSpeedTen = temp*0.1;
  353. 						modbusPojo.setWindSpeedTen(windSpeedTen);
  354. 						break;
  355. 					case 22:
  356. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 22, 23);
  357. 						modbusPojo.setDailyExposure(temp);
  358. 						break;
  359. 					case 24:
  360. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 24, 25);
  361. 						double totalExposure = temp*0.001;
  362. 						modbusPojo.setTotalExposure(totalExposure);
  363. 						break;
  364. 					case 26:
  365. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 26, 27);
  366. 						double scaExposure = temp*0.001;
  367. 						modbusPojo.setScaExposure(scaExposure);
  368. 						break;
  369. 					case 28:
  370. 						temp = ModbusUtil.getUnsignedAns(recvBuff, 28, 29);
  371. 						double dirExposure = temp*0.001;
  372. 						modbusPojo.setDirExposure(dirExposure);
  373. 						break;
  374. 				}
  375. 				i=i+2;
  376. 			}
  377. 		}
  378. 	}
  379. 	
  380. 	public void sendProcess()
  381. 	{
  382. 		byte[] message = new byte[8];
  383. 		int sendLen = 0;
  384. 		message[sendLen++] = devAddr;
  385. 		message[sendLen++] = 0x03;
  386. 		message[sendLen++] = 0x00;
  387. 		message[sendLen++] = 0x00;
  388. 		message[sendLen++] = 0x00;
  389. 		message[sendLen++] = 0x0F;
  390. 		byte[] crc = CommUtil.CRC16(message,6);
  391. 		message[sendLen++] = crc[0];
  392. 		message[sendLen++] = crc[1];
  393. 		comm.writeBuff(message);
  394. 	}
  395.  
  396. }

 

欢迎跳转到本文的原文链接：https://honeypps.com/network/java-udp-comm/

欢迎支持笔者新作：《深入理解Kafka:核心设计与实践原理》和《RabbitMQ实战指南》，同时欢迎关注笔者的微信公众号：朱小厮的博客。

 

 

相关技术文章

最新源码

下载排行榜
点击QQ咨询
开通会员
返回顶部
×
微信扫码支付 元
微信扫码支付
确定支付下载
请使用微信描二维码支付
×

提示信息

×

选择支付方式

  • 微信支付
  • 支付宝付款
确定支付下载