messagelist.java

双向循环队列

package lhb.util.queue;

import java.io.*;
import java.util.*;

/**
 * <p>Title: MessageList</p>
 * <p>Description: 做为输入也输出的队列实现，MessageList是底层与应用层之间的松散的连接通道。
 * @author lhb
 * @since 2008-01-26
 * @version 2.0
 */
public class MessageList
{

    private final static String module = MessageList.class.getName();

    // 共有4个队列：
    private LinkedList reqClientList = new LinkedList();
    private LinkedList reqServerList = new LinkedList();
    private LinkedList resClientList = new LinkedList();
    private LinkedList resServerList = new LinkedList();

    //使用单态模式保证当前JVM中只有一个MessageList实例
    private static MessageList messageList = new MessageList();

    /**
     *
     * @return MessageList
     */
    public static MessageList getInstance()
    {
        return messageList;
    }

    /**
     * 请求客户端：取出reqClientList中第一个数据
     * @return ByteArrayOutputStream
     */
    public ByteArrayOutputStream removeReqFirst()
    {
        synchronized (reqClientList)
        {
            while (reqClientList.isEmpty())
            {
                try
                {
                    //等待解锁 等待加入数据后的提醒
                    reqClientList.wait();
                } catch (InterruptedException ie)
                {
                }
            }
            return (ByteArrayOutputStream) reqClientList.removeFirst();
        }
    }

    /**
     * 请求服务端：加入数据到reqServerList
     * @param in inputStream
     */
    public void pushRequest(InputStream in)
    {
        synchronized (reqServerList)
        {
            reqServerList.add(in);
            // 提醒锁在reqServerList的其它线程
            reqServerList.notifyAll();
        }
    }

    /**
     * 响应客户端：将响应信息放入resClientList
     * @param inputStream InputStream
     */
    public void pushResponse(InputStream inputStream)
    {
        synchronized (resClientList)
        {
            resClientList.add(inputStream);
            // 提醒锁在resClientList的其它线程
            resClientList.notifyAll();
        }
    }

    /**
     * 响应服务端：取resServerList中第一条数据
     * @return ByteArrayOutputStream
     */
    public ByteArrayOutputStream removeResFirst()
    {
        synchronized (resServerList)
        {
            // 如果没有数据，就锁定在这里，防止CPU独霸
            while (resServerList.isEmpty())
            {
                try
                {
                    //等待解锁 等待加入数据后的提醒
                    resServerList.wait();
                } catch (InterruptedException ex)
                {
                    ex.printStackTrace();
                }
            }
            System.out.println("###### resServerList out ######");
            return (ByteArrayOutputStream) resServerList.removeFirst();
        }
    }

    /**
     * 请求服务端：加入数据到reqServerList
     * @param outputStream  OutputStream
     */
    public void pushRequest(OutputStream outputStream)
    {
        synchronized (reqServerList)
        {
            reqServerList.add(outputStream);
            reqServerList.notifyAll();
        }
    }

    /**
     * 响应客户端：将响应信息放入resClientList
     * @param outputStream  OutputStream
     */
    public void pushResponse(OutputStream outputStream)
    {
        synchronized (resClientList)
        {
            resClientList.add(outputStream);
            resClientList.notifyAll();
        }
    }

    /**
     * 请求客户端
     * @param out OutputStream
     */
    public void pushClientRequest(OutputStream out)
    {
        synchronized (reqClientList)
        {
            reqClientList.add(out);
            reqClientList.notifyAll();
        }
    }

    /**
     * 请求服务端
     * @return InputStream
     */
    public InputStream removeReqServerFirst()
    {
        synchronized (reqServerList)
        {
            // 如果没有数据，就锁定在这里，防止CPU独霸
            while (reqServerList.isEmpty())
            {
                try
                {
                    //等待解锁 等待加入数据后的提醒
                    System.out.println("###### reqServerList wait ######");
                    reqServerList.wait();
                } catch (InterruptedException ex)
                {
                    ex.printStackTrace();
                }
            }
            System.out.println("###### reqServerList out ######");
            return (InputStream) reqServerList.removeFirst();
        }
    }

    /**
     * 响应客户端，这里供主线程调用，不能设置wait，否则没人来叫醒
     * 设置超时
     * @return InputStream
     */
    public InputStream removeResClientFirst()
    {
        synchronized (resClientList)
        {

            // 如果没有数据，就锁定在这里，防止CPU独霸
            while (resClientList.isEmpty())
            {
                try
                {
                    //等待解锁 等待加入数据后的提醒
                    System.out.println("###### resClientList wait ######");
                    resClientList.wait();
                } catch (Exception ex)
                {
                    ex.printStackTrace();
                }
            }

//      if(!resClientList.isEmpty()){
            System.out.println("###### resClientList out ######");
            return (InputStream) resClientList.removeFirst();
//      }
//      return null;
        }
    }

    /**
     * 响应服务端
     * @param outputStream OutputStream
     */
    public void pushServerResponse(OutputStream outputStream)
    {
        synchronized (resServerList)
        {
            resServerList.add(outputStream);
//      System.out.println("###### resServerList in ######");
            resServerList.notifyAll();
        }
    }

    /**
     *
     */
    public void notifyQueue()
    {
//    reqClientList.notifyAll();
        reqServerList.notifyAll();
//    resClientList.notifyAll();
        resServerList.notifyAll();
    }
}