udp多线程接收并持久化数据 期货

本文主要是介绍udp多线程接收并持久化数据 期货，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

环境

java：1.7

场景

最近公司开始准备上线期货的数据。目前暂时的处理是打算把接收到的原始数据先保存到txt文件里面。之后再用其他程序去读取这些txt文件来满足业务的需求。

udp

选用udp是因为交易所那边就是使用这个来发送数据，之前我写过一篇udp的基础知识，这次我用来三个线程来接受并持久化数据：
① 主线程 接受数据
②监听线程：当队列里的数据满1000或者接受时间超过2分钟就调用写线程
③写线程：用于把数据写入到txt文件里去。

代码

HQContainer类

package udp;import java.util.concurrent.LinkedBlockingQueue;public enum HQContainer {HQ;private LinkedBlockingQueue<String> records = new LinkedBlockingQueue<String>();public void add (String record){if(records.size()>=1000){System.out.println("------满一千了--------");}records.offer(record);}public String get() {return records.poll();}public int getSize() {return records.size();}public boolean isEmpty(){return records.isEmpty();}
}

上面这段代码是个枚举类，里面主要是LinkedBlockingQueue这个队列用来装数据，使用枚举是因为其实单例的，多线程处理起来非常方便。

客户端代码：

package udp;import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;public class ClientUDP extends Thread{private static final int TIMEOUT = 5000 * 10;//设置接收数据超时时间private static final int MAXSUM = 5;//设置重发数据的次数private static long millis = System.currentTimeMillis();//当前毫秒数private static boolean childFlag = true;public static void main(String[] args) {ClientUDP udp = new ClientUDP();udp.start();byte[] buf = new byte[4096];try {//客户端在9000端口监听接收到的数据
//			InetAddress byName = InetAddress.getByName("localhost");DatagramSocket ds = new DatagramSocket(9888);//建立了一个通信实例（socket）InetAddress loc = InetAddress.getLocalHost();String host = loc.getHostAddress();System.out.println(host);//定义用来发送数据的DatagramPacket实例  //官方：创建一个数据包实例，将指定长度的数据发送给指定端口的指定主键上。
//			DatagramPacket dp_send = new DatagramPacket(str_send.getBytes(), str_send.length(), loc, 3000);//定义用来接收数据的DatagramPacket实例//没有设置端口号，所以是接收数据，要是设置了端口号和主机就是发送数据DatagramPacket dp_receive = new DatagramPacket(buf, buf.length);//数据发向本地3000端口ds.setSoTimeout(TIMEOUT);int tries = 0;boolean receivedResponse = false;//直到接收到数据，或者重发次数达到预定值，则退出循环System.out.println("-----开始接收数据--------");Thread t = new Thread();t.run();while(tries < MAXSUM){try {//接收从服务端发送来的数据ds.receive(dp_receive);//如果接收的数据不是来自目标地址就抛出异常
//					if(!dp_receive.getAddress().equals(byName)){
//						throw new IOException("Received packet from an umknown source");
//					}				//如果接收到数据。则将receivedResponse标志位改为true，从而退出循环receivedResponse = true;if(receivedResponse){//如果收到数据，则打印出来System.out.println("client received data from server：");String str_receive = new String(dp_receive.getData(),0,dp_receive.getLength(), "utf-8");HQContainer.HQ.add(str_receive);System.out.println("HQContainer LinkedBlockingQueue size：" + HQContainer.HQ.getSize());//由于dp_receive在接收了数据之后，其内部消息长度值会变为实际接收的消息的字节数，  //所以这里要将dp_receive的内部消息长度重新置为1024 dp_receive.setLength(buf.length);}else{//如果重发MAXNUM次数据后，仍未获得服务器发送回来的数据，则打印如下信息  System.out.println("No response -- give up.");}} catch (InterruptedIOException e) {//如果接收数据时阻塞超时，重发并减少一次重发次数tries += 1;System.out.println("Time out," + (MAXSUM - tries) + " more tries..." ); 
//					e.printStackTrace();}}ds.close();childFlag = false;} catch (IOException e) {e.printStackTrace();}}@Overridepublic void run() {System.out.println("-----子线程开始监听-----");while(childFlag){if(HQContainer.HQ.getSize()>1000 || (System.currentTimeMillis() - millis) >= 120000l){millis = System.currentTimeMillis();WriteTxt wt = new WriteTxt();wt.run();}}if(!childFlag){WriteTxt wt = new WriteTxt();wt.run();}}
}

上面这段代码是客户端代码；这个类继承了Thread类，在其run()方法里面写了个死循环用来监听队列是否满了或者接受时间超过2分钟，则调用WriteTxt类（其实现了Runnable接口来）来持久化数据。

WriteTxt 类

package udp;import java.io.File;
import java.io.FileOutputStream;
import java.text.SimpleDateFormat;
import java.util.Date;public class WriteTxt implements Runnable {@Overridepublic void run() {try {System.out.println("-----WriteTxt-----");if(!HQContainer.HQ.isEmpty()){File f = new File("");SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");String dateStr = sdf.format(new Date());System.out.println("-----开始写文件啦-----");String file = f.getAbsolutePath() + File.separator +"qihuodata" +File.separator + dateStr + "qihuo-zz.txt";File dataFile = new File(file);if(!dataFile.getParentFile().exists()){dataFile.getParentFile().mkdirs();}FileOutputStream fos = new FileOutputStream(dataFile);while(true){String str = HQContainer.HQ.get();byte[] bytes = str.getBytes();fos.write(bytes, 0, bytes.length);if(HQContainer.HQ.isEmpty()){fos.close();break;}}}} catch (Exception e) {e.printStackTrace();}}
}

上面这个类的作用就是持久化；
我文件命名的方式是qihuodata + 年月日时分秒 + qihuo-zz.txt。
zz:郑州的意思。

时间转换

SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
String dateStr = sdf.format(new Date());

模拟服务器端发送数据

由于公司是固定在某一台服务器上接收数据，并且没有给我开权限，所以我就自己模拟数据，怎样保证模拟的数据是正确的呢？我们可以先在那台服务器上把结果打印出来。
代码如下：

package udp;import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;public class ServiceUDP {public static void main(String[] args) throws InterruptedException {byte[] buf = new byte[4096];try {//服务端在3000端口监听接收数据DatagramSocket ds = new DatagramSocket(3000);//首先建立一个通信（socket）实例//接收从客户端发送过来的数据//创建数据包（没有包含端口和IP，所以是接收）//事实上，其包含远程客户端的IP地址和端口DatagramPacket dp_receive = new DatagramPacket(buf, buf.length);System.out.println("server is on，开始发送数据...");InetAddress byName = InetAddress.getByName("localhost");boolean f = true;while(f){String str_send = "SR811C6900          55.5        55.5        59.5        1.5         1           0.0         0           59.5        186         59.5        59.5        59.5        59.5        59.5        59.5        511.5       59.5        299.5       0.5         20          ";//服务器端接收来自客户端的数据
//				ds.receive(dp_receive);/*System.out.println("server received data from client：");String str_receive = new String(dp_receive.getData(), 0, dp_receive.getLength()) + " from " + dp_receive.getAddress().getHostAddress() + ":" +dp_receive.getPort();System.out.println(str_receive);*///数据发送到客户端3000端口//要发送数据，所以需要指定IP和端口DatagramPacket dp_send = new DatagramPacket(str_send.getBytes(), str_send.length(), byName, 9888);ds.send(dp_send);//由于dp_receive在接收了数据之后，其内部消息长度值会变为实际接收的消息的字节数，  //所以这里要将dp_receive的内部消息长度重新置为1024dp_receive.setLength(buf.length);Thread.sleep(1000 * 2);System.out.println("发送的数据："+ str_send);}ds.close();} catch (IOException e) {e.printStackTrace();}}
}

这样就可以在本机来测试数据。

总结

知道udp是如何发送和接受数据之后，就没什么技术问题啦！整体都是很简单的。
今天又从恒生电子那里接收行情数据，其文档看的真操蛋，真不如交易所给的清晰明了。

这篇关于udp多线程接收并持久化数据 期货的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---