Nettyフレームワークによるシンプルなマルチキャスト

8230 ワード

Java

ゲームの中でよくこのような状況に遭遇します.世界/陣営のチャンネルチャット、あるいはあるプレイヤーが特別なNBの品物を獲得したか、ある活動が始まったか、あるグループまたはすべてのクライアントに通知する必要があります.これらはネットワークプログラミングのマルチキャスト/マルチキャストが必要です.実際には、設計モードのパブリケーション-サブスクリプションモードと似ています.NettyフレームワークのChannelGroupインタフェースは、ニーズを実現するためのいくつかの方法を提供します.いつものように、簡単なコード例を直接使用して説明します.簡単にするために、私たちは最も簡単な世界チャットを実現します.原理は、プレイヤーがチャット情報をサーバに送信し、サーバがすべてのリンクされたクライアントにメッセージを転送することです.
メッセージの転送はprotobufプロトコルを使用し続け、Nettyフレームワークは4.0バージョンを使用します.Nettyとprotobufの使用については、前の記事で詳しく説明しています.
まずpbファイルを定義proto:

package message;
option java_package = "message.request";
option java_outer_classname = "ChatRequest";


message Command {
	
	optional int32 chatType = 1;
	
	optional string content = 2;
}

ChatResponse.これとChatRequest.protoの定義は同じです.ChatResponse.proto:

package message;
option java_package = "message.response";
option java_outer_classname = "ChatResponse";


message ChatMessage {
	
	optional int32 chatType = 1;
	
	optional string content = 2;
}

JavaファイルをプロジェクトTestNettyTcpMulticastのsrc目の下に生成します.
サーバ側:NettyTcpMulticastServer.java

package server;

import message.request.ChatRequest;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;


public class NettyTcpMulticastServer {

	public static void main(String[] args) throws Exception {
	
		EventLoopGroup bossGroup = new NioEventLoopGroup(1);
		EventLoopGroup workGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);
		
		try {
			ServerBootstrap bs = new ServerBootstrap();
			bs.group(bossGroup, workGroup)
				.channel(NioServerSocketChannel.class)
				.childHandler(new ChannelInitializer() {
					
					@Override
					public void initChannel(SocketChannel ch) throws Exception {
						
						ProtobufDecoder protobufDecoder = new ProtobufDecoder(ChatRequest.Command.getDefaultInstance());
						
						//Decoder
						ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(2 * 1024 * 1024, 0, 4, 0, 4));
						ch.pipeline().addLast("protobufDecoder", protobufDecoder);
						
						//Encoder
						ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
						ch.pipeline().addLast("protobufEncoder", new ProtobufEncoder());
						
					
						//handler
						ch.pipeline().addLast(new SocketHandler());
					}
				})
				
				.option(ChannelOption.SO_BACKLOG, 128)
				.childOption(ChannelOption.SO_KEEPALIVE, true);
			
			//Bind and start to accept incoming connections
			ChannelFuture cf = bs.bind(8080).sync();
			
			cf.channel().closeFuture().sync();
		} finally {
			workGroup.shutdownGracefully();
			bossGroup.shutdownGracefully();
		}
	}	
}

server論理処理、SockerHandler.java

package server;

import message.request.ChatRequest;
import message.response.ChatResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

public class SocketHandler extends ChannelInboundHandlerAdapter {

	//channels         channel
	private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
	
	
	//    channel  channels 
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		channels.add(ctx.channel());
		super.channelActive(ctx);
	}
	
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		
		ChatRequest.Command command = (ChatRequest.Command)msg;
		
		ChatResponse.ChatMessage.Builder build = ChatResponse.ChatMessage.newBuilder();
		build.setChatType(command.getChatType())
			 .setContent(command.getContent() + "，         ");
		
		ChatResponse.ChatMessage chatMessage = build.build();
		
		//              
		channels.writeAndFlush(chatMessage);
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		cause.printStackTrace();
		ctx.close();
	}
}

Nettyの各socketリンクはchannelであり、すべてのリンクのクライアントchannelをChannelGroupオブジェクトchannelsに追加するだけで、論理処理でchannelsオブジェクトを呼び出すwriteAndFlushメソッドでメッセージを送信すればよい.ChannelGroupの具体的な使い方については、Nettyの公式ドキュメントを参照してください.
ここでは、2つのクライアント、1つのチャットコンテンツを作成し、サーバを介してすべての接続されたクライアントにコンテンツを転送します.これはNettyTcpMulitcastSendReceiveClientです.java

package client;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;

import message.request.ChatRequest;
import message.response.ChatResponse;


public class NettyTcpMulitcastSendReceiveClient {

	public static void main(String[] args) throws IOException {

		Socket socket = null;  
		DataOutputStream dataOut = null;
		DataInputStream dataIn = null;
		
        try {  
  
        	socket = new Socket("localhost", 8080);  
        	
        	dataOut = new DataOutputStream(socket.getOutputStream());
        	dataIn = new DataInputStream(socket.getInputStream());
             
            ChatRequest.Command.Builder requestBuiler = ChatRequest.Command.newBuilder();
            requestBuiler.setChatType(0)
            			 .setContent("     ");
            ChatRequest.Command Command = requestBuiler.build();
            
            byte[] dataOutBytes = Command.toByteArray();  
            
            //   header
            dataOut.writeInt(dataOutBytes.length);
            
            //   body
            dataOut.write(dataOutBytes);
            dataOut.flush();
            
           
            int bodyLength = dataIn.readInt();
            byte[] bodyBytes = new byte[bodyLength];
            dataIn.read(bodyBytes);
            
            //      ChatResponse.ChatMessage  
            ChatResponse.ChatMessage chatResponse = ChatResponse.ChatMessage.parseFrom(bodyBytes);
            
            System.out.println("chatType: " + chatResponse.getChatType());
            System.out.println("chatContent: " + chatResponse.getContent());
 
        } finally {  
            //      
        	dataIn.close();
        	dataOut.close();  
            socket.close();  
        }  
	}
}

これは別のクライアントが単純にチャット情報を受信しているだけです.NettyTcpMulitcastReceiveClient.java

package client;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.Socket;

import message.response.ChatResponse;


public class NettyTcpMulitcastReceiveClient {
	
	public static void main(String[] args) throws IOException {
		
		Socket socket = null;  
		DataInputStream dataIn = null;
		
        try {  
  
        	socket = new Socket("localhost", 8080);    
        	dataIn = new DataInputStream(socket.getInputStream());
             
            int bodyLength = dataIn.readInt();
            byte[] bodyBytes = new byte[bodyLength];
            dataIn.read(bodyBytes);
            
            //      ChatResponse.ChatMessage  
            ChatResponse.ChatMessage chatResponse = ChatResponse.ChatMessage.parseFrom(bodyBytes);
            
            int chatType = chatResponse.getChatType();
            String chatContent = chatResponse.getContent();
            
            System.out.println("chatType: " + chatType);
            System.out.println("chatContent: " + chatContent);
 
        } finally {  
            //      
        	dataIn.close();
            socket.close();  
        }
	}

}

例はかなり簡単です.プログラムをテストして、まずサービス・エンド・プログラムを起動し、NettyTcpMulitcastReceiveClientを起動します.JAvaはsocketのreadにブロックし、NettyTcpMulitcastSendReceiveClientを再起動してチャットコンテンツを送信します.両方のクライアントプログラムがチャット情報を受信していることがわかります.

認定Big Data勉強記 - 8.1: Kinesis Data Streams

蟻の群のアルゴリズムは最短の経路のコードを求めます