?
? ? @Override
? ? public void run() {
? ? ? ? System.out.println(Thread.currentThread().getName() + "---run");
? ? ? ?
? ? ? ? //连接服务器
? ? ? ? channel.connect(new InetSocketAddress("localhost", 8383), null, new CompletionHandler(){
? ? ? ? ? ? final ByteBuffer readBuffer = ByteBuffer.allocateDirect(1024);
? ? ? ? ? ?
? ? ? ? ? ? @Override
? ? ? ? ? ? public void completed(Void result, Void attachment) {
? ? ? ? ? ? ? ? //连接成功后, 异步调用OS向服务器写一条消息
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? //channel.write(CharsetHelper.encode(CharBuffer.wrap(helper.getWord())));
? ? ? ? ? ? ? ? ? ? writeStringMessage(helper.getWord());
? ? ? ? ? ? ? ? } catch (CharacterCodingException e) {
? ? ? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ?
? ? ? ? ? ? ? ? //helper.sleep();//等待写异步调用完成
? ? ? ? ? ? ? ? readBuffer.clear();
? ? ? ? ? ? ? ? //异步调用OS读取服务器发送的消息
? ? ? ? ? ? ? ? channel.read(readBuffer, null, new CompletionHandler(){
?
? ? ? ? ? ? ? ? ? ? @Override
? ? ? ? ? ? ? ? ? ? public void completed(Integer result, Object attachment) {
? ? ? ? ? ? ? ? ? ? ? ? try{
? ? ? ? ? ? ? ? ? ? ? ? ? ? //异步读取完成后处理
? ? ? ? ? ? ? ? ? ? ? ? ? ? if(result > 0){
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? readBuffer.flip();
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? CharBuffer charBuffer = CharsetHelper.decode(readBuffer);
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? String answer = charBuffer.toString();
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? System.out.println(Thread.currentThread().getName() + "---" + answer);
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? readBuffer.clear();
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? String word = helper.getWord();
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? if(word != null){
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? //异步写
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? //channel.write(CharsetHelper.encode(CharBuffer.wrap(word)));
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? writeStringMessage(word);
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? //helper.sleep();//等待异步操作
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? channel.read(readBuffer, null, this);
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? else{
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? //不想发消息了，主动关闭channel
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? shutdown();
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? ? ? ? ? else{
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? //对方已经关闭channel，自己被动关闭，避免空循环
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? shutdown();
? ? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? ? ? catch(Exception e){
? ? ? ? ? ? ? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ?
? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ?
?
? ? ? ? ? ? ? ? ? ? /**
? ? ? ? ? ? ? ? ? ? * 读取失败处理
? ? ? ? ? ? ? ? ? ? * @param exc
? ? ? ? ? ? ? ? ? ? * @param attachment
? ? ? ? ? ? ? ? ? ? */
? ? ? ? ? ? ? ? ? ? @Override
? ? ? ? ? ? ? ? ? ? public void failed(Throwable exc, Object attachment) {
? ? ? ? ? ? ? ? ? ? ? ? System.out.println("client read failed: " + exc);
? ? ? ? ? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? ? ? ? ? shutdown();
? ? ? ? ? ? ? ? ? ? ? ? } catch (IOException e) {
? ? ? ? ? ? ? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ?
? ? ? ? ? ? ? ? });
? ? ? ? ? ? }
?
? ? ? ? ? ? /**
? ? ? ? ? ? * 连接失败处理
? ? ? ? ? ? * @param exc
? ? ? ? ? ? * @param attachment
? ? ? ? ? ? */
? ? ? ? ? ? @Override
? ? ? ? ? ? public void failed(Throwable exc, Void attachment) {
? ? ? ? ? ? ? ? System.out.println("client connect to server failed: " + exc);
? ? ? ? ? ? ? ?
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? shutdown();
? ? ? ? ? ? ? ? } catch (IOException e) {
? ? ? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }? ? ? ? ?
? ? ? ? });? ?
? ? }
? ?
? ? private void shutdown() throws IOException {
? ? ? ? if(channel != null){
? ? ? ? ? ? channel.close();
? ? ? ? }
? ? ? ?
? ? ? ? latch.countDown();? ? ? ? ? ? ? ? ? ? ? ? ?
? ? }
? ?
? ? /**
? ? * Enqueues a write of the buffer to the channel.
? ? * The call is asynchronous so the buffer is not safe to modify after
? ? * passing the buffer here.
? ? *
? ? * @param buffer the buffer to send to the channel
? ? */
? ? private void writeMessage(final ByteBuffer buffer) {
? ? ? ? boolean threadShouldWrite = fal |