基于java.nio.channels的编程实践-I

简介:

服务端代码

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class NIOSocketServer extends Thread {
	private static final Logger LOG = LoggerFactory
			.getLogger(NIOSocketServer.class);
	private static final String CHARSET = "UTF-8";
	private static final int BUFFER_SIZE = 1024;
	private static final int FAIL_TRY_NUM = 3;

	private Selector selector;
	private ServerSocketChannel ssc;
	private static NIOSocketServer server;

	/**
	 * 程序入口
	 * 
	 * @param args
	 */
	public static void main(String[] args) {
		server = new NIOSocketServer();
		try {
			// server.setDaemon(true);
			server.initServer();
			server.start();
		} catch (Exception e) {
			// 如果出现异常,则直接关闭客户端
			server.stopServer();
			System.exit(1);
		}
	}

	@Override
	public void run() {
		int failNum = 0;
		while (true) {
			try {
				int select = selector.select();
				if (select > 0) {
					Set<SelectionKey> keys = selector.selectedKeys();
					Iterator<SelectionKey> iter = keys.iterator();
					while (iter.hasNext()) {
						SelectionKey key = iter.next();
						if (key.isAcceptable()) {
							doAcceptable(key);
						}
						if (key.isWritable()) {
							doWriteMessage(key);
						}
						if (key.isReadable()) {
							doReadMessage(key);
						}
						if (key.isConnectable()) {
							doConnectable(key);
						}
						iter.remove();
					}
				}
			} catch (Exception e) {
				failNum++;
				if (failNum > FAIL_TRY_NUM) {
					server.stopServer();
				}
			}
		}

	}

	/**
	 * 初始化服务器端程序,开始监听端口
	 * 
	 * @throws IOException
	 */
	private void initServer() throws IOException {
		selector = Selector.open();
		ssc = ServerSocketChannel.open();
		ssc.configureBlocking(false);
		ssc.socket().bind(new InetSocketAddress(2181));
		ssc.register(selector, SelectionKey.OP_ACCEPT);
	}

	/**
	 * 停止服务器
	 * 
	 * @throws IOException
	 */
	private void stopServer() {
		try {
			if (selector != null && selector.isOpen()) {
				selector.close();
			}
			if (ssc != null && ssc.isOpen()) {
				ssc.close();
			}
		} catch (IOException e) {
			LOG.info("关闭服务端失败:" + e.getMessage());
		}
	}

	/**
	 * 对新的客户端连接进行处理
	 * 
	 * @param key
	 * @throws IOException
	 */
	private void doAcceptable(SelectionKey keythrows IOException {
		ServerSocketChannel tmpSsc = (ServerSocketChannel) key.channel();
		SocketChannel ss = tmpSsc.accept();
		ss.configureBlocking(false);
		ss.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);

	}

	/**
	 * 已连接
	 * 
	 * @param key
	 */
	private void doConnectable(SelectionKey key) {
		LOG.info("connect is ok");
	}

	/**
	 * 写消息到客户端
	 * 
	 * @param key
	 * @throws IOException
	 */
	private void doWriteMessage(SelectionKey keythrows Exception {
		SocketChannel sc = (SocketChannel) key.channel();
		ByteBuffer buffer = ByteBuffer.wrap("server write msg to client"
				.getBytes(CHARSET));
		while (buffer.hasRemaining()) {
			sc.write(buffer);
		}
		TimeUnit.SECONDS.sleep(1);
	}

	/**
	 * @param key
	 * @throws IOException
	 */
	private void doReadMessage(SelectionKey keythrows Exception {
		SocketChannel sc = (SocketChannel) key.channel();
		ByteBuffer bb = ByteBuffer.allocate(BUFFER_SIZE);
		int read = sc.read(bb);
		while (read > 0) {
			bb.flip();
			byte[] barr = new byte[bb.limit()];
			bb.get(barr);
			LOG.info("server read msg from client:" + new String(barr, CHARSET));
			bb.clear();
			read = sc.read(bb);
		}
		TimeUnit.SECONDS.sleep(1);
	}

}

客户端代码

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class NIOSocketClient extends Thread {
	private static final Logger LOG = LoggerFactory
			.getLogger(NIOSocketClient.class);
	private static final String CHARSET = "UTF-8";
	private static final int BUFFER_SIZE = 1024;
	private static final int FAIL_TRY_NUM = 3;

	private SocketChannel socketChannel;
	private Selector selector;
	private static NIOSocketClient client;

	/**
	 * 程序入口
	 * 
	 * @param args
	 */
	public static void main(String[] args) {
		client = new NIOSocketClient();
		try {
			client.initClient();
			client.start();
			// client.setDaemon(true);catch (Exception e) {
			// 如果出现异常,则直接关闭客户端
			client.close();
		}
	}

	public void run() {
		int failNum = 0;
		while (true) {
			try {
				writeMessage();
				int select = selector.select();
				System.out.println(select);
				if (select > 0) {
					Set<SelectionKey> keys = selector.selectedKeys();
					Iterator<SelectionKey> iter = keys.iterator();
					while (iter.hasNext()) {
						SelectionKey sk = iter.next();
						if (sk.isReadable()) {
							readMessage(sk);
						}
						iter.remove();
					}
				}
			} catch (Exception e) {
				// 如果出现三次以上的异常,则关闭客户端
				failNum++;
				if (failNum > FAIL_TRY_NUM) {
					client.close();
					System.exit(1);
				}
			}
		}
	}

	public void readMessage(SelectionKey sk) throws Exception,
			UnsupportedEncodingException {
		SocketChannel curSc = (SocketChannel) sk.channel();
		ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
		while (curSc.read(buffer) > 0) {
			buffer.flip();
			LOG.info("read message from server:"new String(buffer.array(), CHARSET));
			buffer.clear();
		}
		TimeUnit.SECONDS.sleep(1);
	}

	public void writeMessage() throws Exception {
		String ss = "client write msg to server";
		ByteBuffer buffer = ByteBuffer.wrap(ss.getBytes(CHARSET));
		while (buffer.hasRemaining()) {
			socketChannel.write(buffer);
		}

		TimeUnit.SECONDS.sleep(1);
	}

	public void initClient() throws IOException, ClosedChannelException {
		InetSocketAddress addr = new InetSocketAddress(2181);
		socketChannel = SocketChannel.open();

		selector = Selector.open();
		socketChannel.configureBlocking(false);
		socketChannel.register(selector, SelectionKey.OP_READ);

		// 连接到server
		socketChannel.connect(addr);

		while (!socketChannel.finishConnect()) {
			LOG.info("check finish connection");
		}
	}

	/**
	 * 停止客户端
	 */
	private void close() {
		try {
			if (selector != null && selector.isOpen()) {
				selector.close();
			}
			if (socketChannel != null && socketChannel.isOpen()) {
				socketChannel.close();
			}
		} catch (IOException e) {
			LOG.info("关闭客户端失败:" + e.getMessage());
		}
	}

}
目录
相关文章
|
1月前
|
Java 程序员
Java编程中的异常处理:从基础到高级
在Java的世界中,异常处理是代码健壮性的守护神。本文将带你从异常的基本概念出发,逐步深入到高级用法,探索如何优雅地处理程序中的错误和异常情况。通过实际案例,我们将一起学习如何编写更可靠、更易于维护的Java代码。准备好了吗?让我们一起踏上这段旅程,解锁Java异常处理的秘密!
|
15天前
|
存储 缓存 Java
Java 并发编程——volatile 关键字解析
本文介绍了Java线程中的`volatile`关键字及其与`synchronized`锁的区别。`volatile`保证了变量的可见性和一定的有序性,但不能保证原子性。它通过内存屏障实现,避免指令重排序,确保线程间数据一致。相比`synchronized`,`volatile`性能更优,适用于简单状态标记和某些特定场景,如单例模式中的双重检查锁定。文中还解释了Java内存模型的基本概念,包括主内存、工作内存及并发编程中的原子性、可见性和有序性。
Java 并发编程——volatile 关键字解析
|
19天前
|
算法 Java 调度
java并发编程中Monitor里的waitSet和EntryList都是做什么的
在Java并发编程中,Monitor内部包含两个重要队列:等待集(Wait Set)和入口列表(Entry List)。Wait Set用于线程的条件等待和协作,线程调用`wait()`后进入此集合,通过`notify()`或`notifyAll()`唤醒。Entry List则管理锁的竞争,未能获取锁的线程在此排队,等待锁释放后重新竞争。理解两者区别有助于设计高效的多线程程序。 - **Wait Set**:线程调用`wait()`后进入,等待条件满足被唤醒,需重新竞争锁。 - **Entry List**:多个线程竞争锁时,未获锁的线程在此排队,等待锁释放后获取锁继续执行。
54 12
|
15天前
|
监控 Java API
探索Java NIO:究竟在哪些领域能大显身手?揭秘原理、应用场景与官方示例代码
Java NIO(New IO)自Java SE 1.4引入,提供比传统IO更高效、灵活的操作,支持非阻塞IO和选择器特性,适用于高并发、高吞吐量场景。NIO的核心概念包括通道(Channel)、缓冲区(Buffer)和选择器(Selector),能实现多路复用和异步操作。其应用场景涵盖网络通信、文件操作、进程间通信及数据库操作等。NIO的优势在于提高并发性和性能,简化编程;但学习成本较高,且与传统IO存在不兼容性。尽管如此,NIO在构建高性能框架如Netty、Mina和Jetty中仍广泛应用。
26 3
|
16天前
|
存储 安全 Java
Java多线程编程秘籍:各种方案一网打尽,不要错过!
Java 中实现多线程的方式主要有四种:继承 Thread 类、实现 Runnable 接口、实现 Callable 接口和使用线程池。每种方式各有优缺点,适用于不同的场景。继承 Thread 类最简单,实现 Runnable 接口更灵活,Callable 接口支持返回结果,线程池则便于管理和复用线程。实际应用中可根据需求选择合适的方式。此外,还介绍了多线程相关的常见面试问题及答案,涵盖线程概念、线程安全、线程池等知识点。
98 2
|
23天前
|
存储 监控 Java
Java的NIO体系
通过本文的介绍,希望您能够深入理解Java NIO体系的核心组件、工作原理及其在高性能应用中的实际应用,并能够在实际开发中灵活运用这些知识,构建高效的Java应用程序。
31 5
|
1月前
|
存储 监控 小程序
Java中的线程池优化实践####
本文深入探讨了Java中线程池的工作原理,分析了常见的线程池类型及其适用场景,并通过实际案例展示了如何根据应用需求进行线程池的优化配置。文章首先介绍了线程池的基本概念和核心参数,随后详细阐述了几种常见的线程池实现(如FixedThreadPool、CachedThreadPool、ScheduledThreadPool等)的特点及使用场景。接着,通过一个电商系统订单处理的实际案例,分析了线程池参数设置不当导致的性能问题,并提出了相应的优化策略。最终,总结了线程池优化的最佳实践,旨在帮助开发者更好地利用Java线程池提升应用性能和稳定性。 ####
|
1月前
|
安全 算法 Java
Java多线程编程中的陷阱与最佳实践####
本文探讨了Java多线程编程中常见的陷阱,并介绍了如何通过最佳实践来避免这些问题。我们将从基础概念入手,逐步深入到具体的代码示例,帮助开发者更好地理解和应用多线程技术。无论是初学者还是有经验的开发者,都能从中获得有价值的见解和建议。 ####
|
1月前
|
Java 调度
Java中的多线程编程与并发控制
本文深入探讨了Java编程语言中多线程编程的基础知识和并发控制机制。文章首先介绍了多线程的基本概念,包括线程的定义、生命周期以及在Java中创建和管理线程的方法。接着,详细讲解了Java提供的同步机制,如synchronized关键字、wait()和notify()方法等,以及如何通过这些机制实现线程间的协调与通信。最后,本文还讨论了一些常见的并发问题,例如死锁、竞态条件等,并提供了相应的解决策略。
51 3
|
1月前
|
安全 Java 数据库连接
Java中的异常处理:理解与实践
在Java的世界里,异常处理是维护代码健壮性的守门人。本文将带你深入理解Java的异常机制,通过直观的例子展示如何优雅地处理错误和异常。我们将从基本的try-catch结构出发,探索更复杂的finally块、自定义异常类以及throw关键字的使用。文章旨在通过深入浅出的方式,帮助你构建一个更加稳定和可靠的应用程序。
34 5