线程池实现socket 通信问题，不能长时间运行。

0 0

线程池实现socket 通信问题，不能长时间运行。3

1.创建一个监听

import java.io.IOException;
import java.net.*;
import java.util.concurrent.Future;

public class Listener extends ServerSocket implements Runnable {

	public Listener() throws IOException {
		super(Server.AppConfig.getLocalhost().getListenport());
	}

	@Override
	public void run() {
		while (true) {
			try {
				Socket socket = accept();
				CreateServer server = new CreateServer(socket, Server.pool);
				Future<Integer> result = Server.pool.submit(server);				
				Server.Results.add(result);
			} catch (Exception e) {
				Server.createMessage("Listener:"+e.getMessage());
			} finally {
			}
		}

	}
}

2. 创建一个解析socket的服务

import java.net.Socket;
import java.util.Date;
import java.util.Scanner;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.io.*;

import com.szcust.einfo.receiveEntity.serverEntity.DataText;

public class CreateServer implements  Callable<Integer> {
	private Socket client;
	private Scanner in;
	private PrintWriter out;
	private Resolve resolve;
	private int timeOut = 30;
	private Date lastTime;

	public CreateServer(Socket s, ExecutorService pool) {
		client = s;
		lastTime = new Date();
		resolve = new Resolve();
		try {
			client.setSoTimeout(30 * 60 * 1000);
			Server.ClientCount = Server.ClientCount + 1;
			in = new Scanner(client.getInputStream(), "GB2312");

			// in = new BufferedReader(new
			// InputStreamReader(client.getInputStream(), "GB2312"));
			out = new PrintWriter(client.getOutputStream(), true);
			out
					.println("--- Welcome To Universtar Science & Technology Softwear System ---");

		} catch (Exception ex) {
			Server.createMessage("Ex " + ex.getMessage());
		}
	}

	@Override
	public Integer call() {
		String line = "";
		while ((line = in.next()) != null) {
			try {
				if (check(line)) {
					DataText dataText = resolve.getDataTextBySoketString(line);
					if (dataText != null) {
						// Server.Data_Array.add(dataText);
						resolve.saveDataRun(dataText);//业务代码
					} else {
						Server.createMessage("Resolve error " + line);
					}
				} else {
					Server.createMessage("Check error " + line);
				}

			} catch (Exception ex) {
				Server.createMessage("Ex " + ex.getMessage());
				closeSocket(this.client	);
			}
		}
		Server.ClientCount--;
		return Server.ClientCount;
	}

3. 静态变量，用于保存各个线程之间的数据。

import java.util.Calendar;
import java.util.Vector;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import com.szcust.einfo.receiveBiz.FactorBiz;
import com.szcust.einfo.receiveBiz.StationBiz;
import com.szcust.einfo.receiveEntity.einfoEntity.Station;
import com.szcust.einfo.receiveEntity.serverEntity.DataText;
import com.szcust.einfo.receiveEntity.configEntity.Config;
import com.szcust.einfo.receiveEntity.configEntity.ExLog;

public class Server{


	public static  ConcurrentLinkedQueue<DataText> Data_Array = null;
	public static Vector<Station> Client_Stations = null;
	public static Vector<Station> Server_Stations = null;
	private static ConcurrentLinkedQueue<String> Message = null;
	public static ConcurrentLinkedQueue<String> ErrorText = null;
	public static ConcurrentLinkedQueue<Future<Integer>> Results = null;

	public static ExecutorService pool = Executors.newCachedThreadPool();
	public static Config AppConfig = null;
	public static int ClientCount = 0;
	public static boolean IsClear = true;

	public static void  init() {
		Data_Array =new ConcurrentLinkedQueue<DataText>();
		Client_Stations = new Vector<Station>();
		Server_Stations = new Vector<Station>();
		Message = new ConcurrentLinkedQueue<String>();
		ErrorText = new ConcurrentLinkedQueue<String>();
		Results = new ConcurrentLinkedQueue<Future<Integer>>();
		AppConfig = new Config();
		ClientCount = 0;
		IsClear = true;
		loadData();
	}

问题补充：

dwangel 写道

Blocking Socket不适合用线程池处理通讯。
单个线程会被挂起的。
用NIO才合适。

我这个是个长连接，还有心跳包不断的发上来。

我对那个NIO 不是很熟，但是我试用到时候好像有新的连接时要断开原有的连接，不能长连接。

期待解决！！

问题补充：

dwangel 写道

Blocking Socket不适合用线程池处理通讯。
单个线程会被挂起的。
用NIO才合适。

yx200404 写道

个人觉得没必要在accept后submit一个Callable

直接execute一个Runnable

同步好ClientCount就可以了吧

我仔细检查并写日志调试了，上面的好像没有问题。但是问题归于：

package com.szcust.einfo.receive;

import java.util.ArrayList;

import com.szcust.einfo.receiveBiz.BaseDataBiz;
import com.szcust.einfo.receiveEntity.serverEntity.BaseData;

public class DataSave extends Thread {
	private Resolve resolve;
	private BaseDataBiz baseDataBiz;

	public DataSave() {
		resolve = new Resolve();
		baseDataBiz = new BaseDataBiz();
	}

	@Override

	public void run() {
		while (true) {
			try {
				if (dataArrayCheck(Server.Data_Array)) {
					resolve.saveDataRun(Server.Data_Array.poll());
				} else {
					this.setPriority(MAX_PRIORITY);
					Thread.sleep(10 * 1000);
				}
			} catch (Exception ex) {
				Server.createMessage("DataSave: " + ex.getMessage());
			}
		}
	}

一直打印空错误：DataSave:null

问题补充：

dwangel 写道

你的项目可以用外部库吗？

你可以看看Netty这个NIO库。(或者apache mina，netty是mina的创始者写的，我用起来蛮方便的。)

NIO是基于SELECT模型的。就是注册一批关注对象socketChannel，
然后重复调用Select方法，当SocketChannel处于指定状态，比如有数据，可写
等状态，再拿出来处理。
这样，可以分成主线程持续select,select出来需要处理的socketchannel放到一个队列，由threadpool里分线程取出来处理。

因为socketchannel操作不会阻塞，所以可以充分利用多线程。

而你的代码用的是普通socket，普通socket的特性就是在read时，如果没有数据，则会阻塞住。而如果想利用线程池，应当是read没有数据时，继续往下走，把当前线程资源还回线程池。

我开始用了那个mina 但是我发现我的连接有很多，他开了很多线程。但是我不知道怎么及时的结束这些线程。

import java.net.InetSocketAddress;
import java.nio.charset.Charset;

import org.apache.mina.common.DefaultIoFilterChainBuilder;
import org.apache.mina.common.IoAcceptor;
import org.apache.mina.common.IoAcceptorConfig;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.SocketAcceptor;
import org.apache.mina.transport.socket.nio.SocketAcceptorConfig;

public class Service implements Runnable {

	private IoAcceptor acceptor;
	private IoAcceptorConfig config;

	public Service() {
		acceptor = new SocketAcceptor();
		config = new SocketAcceptorConfig();
		DefaultIoFilterChainBuilder chain = config.getFilterChain();
		// 使用字符串编码
		chain.addLast("codec", new ProtocolCodecFilter(
				new TextLineCodecFactory(Charset.forName( "UTF-8" ))));

	}

	public void run() {
		try{
		acceptor.bind(new InetSocketAddress(Server.AppConfig.getLocalhost()
				.getListenport()), new MinaHandler(), config);
		}catch (Exception e) {
			Server.createMessage("Server Ex: "+e.getMessage());
		}
	}

}



import org.apache.mina.common.IoHandlerAdapter;
import org.apache.mina.common.IoSession;

import com.szcust.einfo.receiveEntity.serverEntity.DataText;


public class MinaHandler  extends IoHandlerAdapter {
	/**
	 * 当有异常发生时触发
	 */
	public void exceptionCaught(IoSession ssn, Throwable cause) {
		//cause.printStackTrace();
		Server.createMessage("Handler Ex : " + cause.getMessage());
		ssn.getCloseFuture().getSession().close();
		ssn.close();
	}

	/**
	 * 有新连接时触发
	 */
	@Override
	public void sessionOpened(IoSession ssn) throws Exception {
		ssn.write("--- Welcome To Universtar Science & Technology Softwear System ---");
		//System.out.println("session open for " + ssn.getRemoteAddress());
	}

	/**
	 * 连接被关闭时触发
	 */
	@Override
	public void sessionClosed(IoSession ssn) throws Exception {
		ssn.close();
		//System.out.println("session closed from " + ssn.getRemoteAddress());		
	}

	/**
	 * 收到来自客户端的消息
	 */
	public void messageReceived(IoSession ssn, Object msg) throws Exception {
		//String ip = ssn.getRemoteAddress().toString();		
		//ssn.write("Hello " + msg);
		Resolve resolve = new Resolve();
		DataText dataText = resolve.getDataTextBySoketString(msg.toString());
		if (dataText != null) {
			// Server.Data_Array.add(dataText);
			resolve.saveDataRun(dataText);
		} else {
			Server.createMessage("Resolve error : " + msg);
		}
	}
}

问题补充：我对这个方面是第一次接触，期待解决。

问题补充：

dwangel 写道

不用你主动结束那些线程，那些线程被应该会被设为 daemon=true，即主线程结束时自动结束。平时，等待主监听线程传需要处理的任务进来，如果没有，则自己wait。

那些就是线程池里的线程。

我的是长连接，现在是每读取一行字符串，他都生成一个线程。