MetaQ技术内幕——源码分析(七)

wbj0110

浏览: 1610164 次
性别:
来自: 上海

最近访客更多访客>>

一往无前bhz

ninja2006

loginboot

u012363178

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

MetaQ

MetaQ

前面介绍了Broker在网络传输过程中使用的数据结构，同时也介绍了MetaQ使用了Gecko框架作为网络传输框架。

有人会问，Gecko什么调用MetaEncodeCommand的encode()方法，让命令变成可见的明文在网络传输，Gecko又在什么时候将网络传输的数据包装成一个个Command对象？

或许有人已经注意到了笔者在介绍Broker启动类MetaMorphosisBroker的时候估计漏掉了一个方法newRemotingServer()方法，即创建Gecko Server。

Java代码

private static RemotingServer newRemotingServer(final MetaConfig metaConfig) {
final ServerConfig serverConfig = new ServerConfig();
serverConfig.setWireFormatType(new MetamorphosisWireFormatType()); //注册了MetamorphosisWireFormatType实例，该实例负责编码和解码Command
serverConfig.setPort(metaConfig.getServerPort());
final RemotingServer server = RemotingFactory.newRemotingServer(serverConfig);
return server;
}

private static RemotingServer newRemotingServer(final MetaConfig metaConfig) {		final ServerConfig serverConfig = new ServerConfig();		serverConfig.setWireFormatType(new MetamorphosisWireFormatType()); //注册了MetamorphosisWireFormatType实例，该实例负责编码和解码Command		serverConfig.setPort(metaConfig.getServerPort());		final RemotingServer server = RemotingFactory.newRemotingServer(serverConfig);		return server;}

在该方法内注册了一个MetamorphosisWireFormatType实例，该实例负责Command 的编码解码工作，MetamorphosisWireFormatType实现接口WireFormatType。

Java代码

public class MetamorphosisWireFormatType extends WireFormatType {
public static final String SCHEME = "meta";
public String getScheme() {
return SCHEME;
}
public String name() {
return "metamorphosis";
}
public CodecFactory newCodecFactory() {
return new MetaCodecFactory();
}
public CommandFactory newCommandFactory() {
return new MetaCommandFactory();
}

public class MetamorphosisWireFormatType extends WireFormatType {	public static final String SCHEME = "meta";	public String getScheme() {		return SCHEME;	}	public String name() {		return "metamorphosis";	}	public CodecFactory newCodecFactory() {		return new MetaCodecFactory();	}	public CommandFactory newCommandFactory() {		return new MetaCommandFactory();	}

MetamorphosisWireFormatType本身并没有进行编码解码，而是交给了类MetaCodecFactory去实现，另外我们也看到newCommandFactory()方法，该方法主要是用于连接的心跳检测。下面让我们分别来看看这两个类: MetaCommandFactory和MetaCodecFactory，MetaCommandFactory和MetaCodecFactory均是MetamorphosisWireFormatType的内部类

用于心跳检测的类MetaCommandFactory，该类主要有两个方法，创建心跳请求的createHeartBeatCommand()方法和响应心跳请求的createBooleanAckCommand()方法：

Java代码

static class MetaCommandFactory implements CommandFactory {
public BooleanAckCommand createBooleanAckCommand(final CommandHeader request, final ResponseStatus responseStatus, final String errorMsg) {
//响应心跳请求
int httpCode = -1;
switch (responseStatus) {
case NO_ERROR:
httpCode = HttpStatus.Success;
break;
case THREADPOOL_BUSY:
case NO_PROCESSOR:
httpCode = HttpStatus.ServiceUnavilable;
break;
case TIMEOUT:
httpCode = HttpStatus.GatewayTimeout;
break;
default:
httpCode = HttpStatus.InternalServerError;
break;
}
return new BooleanCommand(httpCode, errorMsg, request.getOpaque());
}
public HeartBeatRequestCommand createHeartBeatCommand() {
//前面介绍过VersionCommand用于心跳检测，就是用于此处
return new VersionCommand(OpaqueGenerator.getNextOpaque());
}
}

static class MetaCommandFactory implements CommandFactory {		public BooleanAckCommand createBooleanAckCommand(final CommandHeader request, final ResponseStatus responseStatus, final String errorMsg) {//响应心跳请求			int httpCode = -1;			switch (responseStatus) {				case NO_ERROR:					httpCode = HttpStatus.Success;					break;				case THREADPOOL_BUSY:				case NO_PROCESSOR:					httpCode = HttpStatus.ServiceUnavilable;					break;				case TIMEOUT:					httpCode = HttpStatus.GatewayTimeout;					break;				default:					httpCode = HttpStatus.InternalServerError;					break;			}			return new BooleanCommand(httpCode, errorMsg, request.getOpaque());		}		public HeartBeatRequestCommand createHeartBeatCommand() {//前面介绍过VersionCommand用于心跳检测，就是用于此处			return new VersionCommand(OpaqueGenerator.getNextOpaque());		}	}

MetaCodecFactory是MetaQ（包括Broker和Client，因为编码解码Broker和Client都需要）网络传输最重要的一个类，负责命令的编码解码，MetaCodecFactory要实现Gecko框架定义的接口CodecFactory，MetaCodecFactory实例才能被Gecko框架使用，接口CodecFactory就定义了两个方法，返回编码器和解码器（由于Client和Broker均需要使用到MetamorphosisWireFormatType，所以MetamorphosisWireFormatType放在common工程中）：

Java代码

static class MetaCodecFactory implements CodecFactory {
//返回解码器
@Override
public Decoder getDecoder() {
return new Decoder() {
//Gecko框架会在适当的时候调用该方法，并将数据放到参数buff中，
//用户可以根据buff的内容进行解析，包装成对应的Command类型
public Object decode(final IoBuffer buff, final Session session) {
if (buff == null || !buff.hasRemaining()) {
return null;
}
buff.mark();
//匹配第一个{‘\r’, ‘\n’}，也就是找到命令的内容(不包括数据)，目前只有PutCommand和SynCommand有数据部分，其他的命令都只有命令的内容
final int index = LINE_MATCHER.matchFirst(buff);
if (index >= 0) {
//获取命令内容
final byte[] bytes = new byte[index - buff.position()];
buff.get(bytes);
//跳过\r\n
buff.position(buff.position() + 2);
//将命令字节数组转换成字符串
final String line = ByteUtils.getString(bytes);
if (log.isDebugEnabled()) {
log.debug("Receive command:" + line);
}
//以空格为单位分离内容
final String[] sa = SPLITER.split(line);
if (sa == null || sa.length == 0) {
throw new MetaCodecException("Blank command line.");
}
//判断内容的第一个字母
final byte op = (byte) sa[0].charAt(0);
switch (op) {
case 'p':
//如果是p的话，认为是put命令，具体见MetaEncodeCommand定义的命令的内容并解析put命令，具体格式在每个命令的实现类里的注释都有，下面的各个方法的注释也有部分
return this.decodePut(buff, sa);
case 'g':
//如果是g的话，认为是get命令
return this.decodeGet(sa);
case 't':
//如果是g的话，认为是事务命令
return this.decodeTransaction(sa);
case 'r':
//如果是g的话，认为是结果响应
return this.decodeBoolean(buff, sa);
case 'v':
//如果是v的话，则可能是心跳请求或者数据响应，所以得使用更详细的信息进行判断
if (sa[0].equals("value")) {
return this.decodeData(buff, sa);
} else {
return this.decodeVersion(sa);
}
case 's':
//如果是s的话，则可能是统计请求或者同步，所以得使用更详细的信息进行判断
if (sa[0].equals("stats")) {
return this.decodeStats(sa);
} else {
return this.decodeSync(buff, sa);
}
case 'o':
//如果是o的话，查询最近可用位置请求
return this.decodeOffset(sa);
case 'q':
//如果是q的话，退出连接请求
return this.decodeQuit();
default:
throw new MetaCodecException("Unknow command:" + line);
}
} else {
return null;
}
}
private Object decodeQuit() {
return new QuitCommand();
}
private Object decodeVersion(final String[] sa) {
if (sa.length >= 2) {
return new VersionCommand(Integer.parseInt(sa[1]));
} else {
return new VersionCommand(Integer.MAX_VALUE);
}
}
// offset topic group partition offset opaque\r\n
private Object decodeOffset(final String[] sa) {
this.assertCommand(sa[0], "offset");
return new OffsetCommand(sa[1], sa[2], Integer.parseInt(sa[3]), Long.parseLong(sa[4]), Integer.parseInt(sa[5]));
}
// stats item opaque\r\n
// opaque可以为空
private Object decodeStats(final String[] sa) {
this.assertCommand(sa[0], "stats");
int opaque = Integer.MAX_VALUE;
if (sa.length >= 3) {
opaque = Integer.parseInt(sa[2]);
}
String item = null;
if (sa.length >= 2) {
item = sa[1];
}
return new StatsCommand(opaque, item);
}
// value totalLen opaque\r\n data
private Object decodeData(final IoBuffer buff, final String[] sa) {
this.assertCommand(sa[0], "value");
final int valueLen = Integer.parseInt(sa[1]);
if (buff.remaining() < valueLen) {
buff.reset();
return null;
} else {
final byte[] data = new byte[valueLen];
buff.get(data);
return new DataCommand(data, Integer.parseInt(sa[2]));
}
}
/**
* result code length opaque\r\n message
*
* @param buff
* @param sa
* @return
*/
private Object decodeBoolean(final IoBuffer buff, final String[] sa) {
this.assertCommand(sa[0], "result");
final int valueLen = Integer.parseInt(sa[2]);
if (valueLen == 0) {
return new BooleanCommand(Integer.parseInt(sa[1]), null, Integer.parseInt(sa[3]));
} else {
if (buff.remaining() < valueLen) {
buff.reset();
return null;
} else {
final byte[] data = new byte[valueLen];
buff.get(data);
return new BooleanCommand(Integer.parseInt(sa[1]), ByteUtils.getString(data), Integer.parseInt(sa[3]));
}
}
}
// get topic group partition offset maxSize opaque\r\n
private Object decodeGet(final String[] sa) {
this.assertCommand(sa[0], "get");
return new GetCommand(sa[1], sa[2], Integer.parseInt(sa[3]), Long.parseLong(sa[4]), Integer.parseInt(sa[5]), Integer.parseInt(sa[6]));
}
// transaction key sessionId type [timeout] [unique qualifier]
// opaque\r\n
private Object decodeTransaction(final String[] sa) {
this.assertCommand(sa[0], "transaction");
final TransactionId transactionId = this.getTransactionId(sa[1]);
final TransactionType type = TransactionType.valueOf(sa[3]);
switch (sa.length) {
case 7:
// Both include timeout and unique qualifier.
int timeout = Integer.valueOf(sa[4]);
String uniqueQualifier = sa[5];
TransactionInfo info = new TransactionInfo(transactionId, sa[2], type, uniqueQualifier, timeout);
return new TransactionCommand(info, Integer.parseInt(sa[6]));
case 6:
// Maybe timeout or unique qualifier
if (StringUtils.isNumeric(sa[4])) {
timeout = Integer.valueOf(sa[4]);
info = new TransactionInfo(transactionId, sa[2], type, null, timeout);
return new TransactionCommand(info, Integer.parseInt(sa[5]));
} else {
uniqueQualifier = sa[4];
info = new TransactionInfo(transactionId, sa[2], type, uniqueQualifier, 0);
return new TransactionCommand(info, Integer.parseInt(sa[5]));
}
case 5:
// Without timeout and unique qualifier.
info = new TransactionInfo(transactionId, sa[2], type, null);
return new TransactionCommand(info, Integer.parseInt(sa[4]));
default:
throw new MetaCodecException("Invalid transaction command:" + StringUtils.join(sa));
}
}
private TransactionId getTransactionId(final String s) {
return TransactionId.valueOf(s);
}
// sync topic partition value-length flag msgId
// opaque\r\n
private Object decodeSync(final IoBuffer buff, final String[] sa) {
this.assertCommand(sa[0], "sync");
final int valueLen = Integer.parseInt(sa[3]);
if (buff.remaining() < valueLen) {
buff.reset();
return null;
} else {
final byte[] data = new byte[valueLen];
buff.get(data);
switch (sa.length) {
case 7:
// old master before 1.4.4
return new SyncCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Long.valueOf(sa[5]), -1, Integer.parseInt(sa[6]));
case 8:
// new master since 1.4.4
return new SyncCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Long.valueOf(sa[5]), Integer.parseInt(sa[6]), Integer.parseInt(sa[7]));
default:
throw new MetaCodecException("Invalid Sync command:" + StringUtils.join(sa));
}
}
}
// put topic partition value-length flag checksum
// [transactionKey]
// opaque\r\n
private Object decodePut(final IoBuffer buff, final String[] sa) {
this.assertCommand(sa[0], "put");
final int valueLen = Integer.parseInt(sa[3]);
if (buff.remaining() < valueLen) {
buff.reset();
return null;
} else {
final byte[] data = new byte[valueLen];
buff.get(data);
switch (sa.length) {
case 6:
// old clients before 1.4.4
return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, null, Integer.parseInt(sa[4]), Integer.parseInt(sa[5]));
case 7:
// either transaction command or new clients since
// 1.4.4
String slot = sa[5];
char firstChar = slot.charAt(0);
if (Character.isDigit(firstChar) || '-' == firstChar) {
// slot is checksum.
int checkSum = Integer.parseInt(slot);
return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), checkSum, null, Integer.parseInt(sa[6]));
} else {
// slot is transaction id.
return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, this.getTransactionId(slot), Integer.parseInt(sa[4]), Integer.parseInt(sa[6]));
}
case 8:
// New clients since 1.4.4
// A transaction command
return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Integer.parseInt(sa[5]), this.getTransactionId(sa[6]), Integer.parseInt(sa[7]));
default:
throw new MetaCodecException("Invalid put command:" + StringUtils.join(sa));
}
}
}
private void assertCommand(final String cmd, final String expect) {
if (!expect.equals(cmd)) {
throw new MetaCodecException("Expect " + expect + ",but was " + cmd);
}
}
};
}
@Override
public Encoder getEncoder() {
//返回编码器
return new Encoder() {
@Override
public IoBuffer encode(final Object message, final Session session) {
//框架会在适当的时候调用编码器的encode()方法，前面说过如果响应的命令是DataCommand的时候假设不是zeroCopy的话，会出现问题。原因就在这里，因为如果不使用zeroCopy的话，返回给Gecko框架的是一个DataCommand的实例，这时候会调用到此方法，而此方法并没有按照DataCommand的格式进行编码，解码器会识别不了，所以容易出问题
return ((MetaEncodeCommand) message).encode();
}
};
}

static class MetaCodecFactory implements CodecFactory {	//返回解码器		@Override		public Decoder getDecoder() {			return new Decoder() {				//Gecko框架会在适当的时候调用该方法，并将数据放到参数buff中，                //用户可以根据buff的内容进行解析，包装成对应的Command类型				public Object decode(final IoBuffer buff, final Session session) {					if (buff == null || !buff.hasRemaining()) {						return null;					}					buff.mark();                     //匹配第一个{‘\r’, ‘\n’}，也就是找到命令的内容(不包括数据)，目前只有PutCommand和SynCommand有数据部分，其他的命令都只有命令的内容					final int index = LINE_MATCHER.matchFirst(buff);					if (index >= 0) {                          //获取命令内容						final byte[] bytes = new byte[index - buff.position()];						buff.get(bytes);						//跳过\r\n						buff.position(buff.position() + 2);                          //将命令字节数组转换成字符串						final String line = ByteUtils.getString(bytes);						if (log.isDebugEnabled()) {							log.debug("Receive command:" + line);						}                          //以空格为单位分离内容						final String[] sa = SPLITER.split(line);						if (sa == null || sa.length == 0) {							throw new MetaCodecException("Blank command line.");						}                          //判断内容的第一个字母						final byte op = (byte) sa[0].charAt(0);						switch (op) {							case 'p':                                    //如果是p的话，认为是put命令，具体见MetaEncodeCommand定义的命令的内容并解析put命令，具体格式在每个命令的实现类里的注释都有，下面的各个方法的注释也有部分								return this.decodePut(buff, sa);							case 'g':                                   //如果是g的话，认为是get命令								return this.decodeGet(sa);							case 't':                                   //如果是g的话，认为是事务命令								return this.decodeTransaction(sa);							case 'r':                                   //如果是g的话，认为是结果响应								return this.decodeBoolean(buff, sa);							case 'v':                                     //如果是v的话，则可能是心跳请求或者数据响应，所以得使用更详细的信息进行判断								if (sa[0].equals("value")) {									return this.decodeData(buff, sa);								} else {									return this.decodeVersion(sa);								}							case 's':							   //如果是s的话，则可能是统计请求或者同步，所以得使用更详细的信息进行判断if (sa[0].equals("stats")) {									return this.decodeStats(sa);								} else {									return this.decodeSync(buff, sa);								}							case 'o':                                  //如果是o的话，查询最近可用位置请求								return this.decodeOffset(sa);							case 'q':                                   //如果是q的话，退出连接请求								return this.decodeQuit();							default:								throw new MetaCodecException("Unknow command:" + line);						}					} else {						return null;					}				}				private Object decodeQuit() {					return new QuitCommand();				}				private Object decodeVersion(final String[] sa) {					if (sa.length >= 2) {						return new VersionCommand(Integer.parseInt(sa[1]));					} else {						return new VersionCommand(Integer.MAX_VALUE);					}				}				// offset topic group partition offset opaque\r\n				private Object decodeOffset(final String[] sa) {					this.assertCommand(sa[0], "offset");					return new OffsetCommand(sa[1], sa[2], Integer.parseInt(sa[3]), Long.parseLong(sa[4]), Integer.parseInt(sa[5]));				}				// stats item opaque\r\n				// opaque可以为空				private Object decodeStats(final String[] sa) {					this.assertCommand(sa[0], "stats");					int opaque = Integer.MAX_VALUE;					if (sa.length >= 3) {						opaque = Integer.parseInt(sa[2]);					}					String item = null;					if (sa.length >= 2) {						item = sa[1];					}					return new StatsCommand(opaque, item);				}				// value totalLen opaque\r\n data				private Object decodeData(final IoBuffer buff, final String[] sa) {					this.assertCommand(sa[0], "value");					final int valueLen = Integer.parseInt(sa[1]);					if (buff.remaining() < valueLen) {						buff.reset();						return null;					} else {						final byte[] data = new byte[valueLen];						buff.get(data);						return new DataCommand(data, Integer.parseInt(sa[2]));					}				}				/**				 * result code length opaque\r\n message				 * 				 * @param buff				 * @param sa				 * @return				 */				private Object decodeBoolean(final IoBuffer buff, final String[] sa) {					this.assertCommand(sa[0], "result");					final int valueLen = Integer.parseInt(sa[2]);					if (valueLen == 0) {						return new BooleanCommand(Integer.parseInt(sa[1]), null, Integer.parseInt(sa[3]));					} else {						if (buff.remaining() < valueLen) {							buff.reset();							return null;						} else {							final byte[] data = new byte[valueLen];							buff.get(data);							return new BooleanCommand(Integer.parseInt(sa[1]), ByteUtils.getString(data), Integer.parseInt(sa[3]));						}					}				}				// get topic group partition offset maxSize opaque\r\n				private Object decodeGet(final String[] sa) {					this.assertCommand(sa[0], "get");					return new GetCommand(sa[1], sa[2], Integer.parseInt(sa[3]), Long.parseLong(sa[4]), Integer.parseInt(sa[5]), Integer.parseInt(sa[6]));				}				// transaction key sessionId type [timeout] [unique qualifier]				// opaque\r\n				private Object decodeTransaction(final String[] sa) {					this.assertCommand(sa[0], "transaction");					final TransactionId transactionId = this.getTransactionId(sa[1]);					final TransactionType type = TransactionType.valueOf(sa[3]);					switch (sa.length) {						case 7:							// Both include timeout and unique qualifier.							int timeout = Integer.valueOf(sa[4]);							String uniqueQualifier = sa[5];							TransactionInfo info = new TransactionInfo(transactionId, sa[2], type, uniqueQualifier, timeout);							return new TransactionCommand(info, Integer.parseInt(sa[6]));						case 6:							// Maybe timeout or unique qualifier							if (StringUtils.isNumeric(sa[4])) {								timeout = Integer.valueOf(sa[4]);								info = new TransactionInfo(transactionId, sa[2], type, null, timeout);								return new TransactionCommand(info, Integer.parseInt(sa[5]));							} else {								uniqueQualifier = sa[4];								info = new TransactionInfo(transactionId, sa[2], type, uniqueQualifier, 0);								return new TransactionCommand(info, Integer.parseInt(sa[5]));							}						case 5:							// Without timeout and unique qualifier.							info = new TransactionInfo(transactionId, sa[2], type, null);							return new TransactionCommand(info, Integer.parseInt(sa[4]));						default:							throw new MetaCodecException("Invalid transaction command:" + StringUtils.join(sa));					}				}				private TransactionId getTransactionId(final String s) {					return TransactionId.valueOf(s);				}				// sync topic partition value-length flag msgId				// opaque\r\n				private Object decodeSync(final IoBuffer buff, final String[] sa) {					this.assertCommand(sa[0], "sync");					final int valueLen = Integer.parseInt(sa[3]);					if (buff.remaining() < valueLen) {						buff.reset();						return null;					} else {						final byte[] data = new byte[valueLen];						buff.get(data);						switch (sa.length) {							case 7:								// old master before 1.4.4								return new SyncCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Long.valueOf(sa[5]), -1, Integer.parseInt(sa[6]));							case 8:								// new master since 1.4.4								return new SyncCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Long.valueOf(sa[5]), Integer.parseInt(sa[6]), Integer.parseInt(sa[7]));							default:								throw new MetaCodecException("Invalid Sync command:" + StringUtils.join(sa));						}					}				}				// put topic partition value-length flag checksum				// [transactionKey]				// opaque\r\n				private Object decodePut(final IoBuffer buff, final String[] sa) {					this.assertCommand(sa[0], "put");					final int valueLen = Integer.parseInt(sa[3]);					if (buff.remaining() < valueLen) {						buff.reset();						return null;					} else {						final byte[] data = new byte[valueLen];						buff.get(data);						switch (sa.length) {							case 6:								// old clients before 1.4.4								return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, null, Integer.parseInt(sa[4]), Integer.parseInt(sa[5]));							case 7:								// either transaction command or new clients since								// 1.4.4								String slot = sa[5];								char firstChar = slot.charAt(0);								if (Character.isDigit(firstChar) || '-' == firstChar) {									// slot is checksum.									int checkSum = Integer.parseInt(slot);									return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), checkSum, null, Integer.parseInt(sa[6]));								} else {									// slot is transaction id.									return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, this.getTransactionId(slot), Integer.parseInt(sa[4]), Integer.parseInt(sa[6]));								}							case 8:								// New clients since 1.4.4								// A transaction command								return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Integer.parseInt(sa[5]), this.getTransactionId(sa[6]), Integer.parseInt(sa[7]));							default:								throw new MetaCodecException("Invalid put command:" + StringUtils.join(sa));						}					}				}				private void assertCommand(final String cmd, final String expect) {					if (!expect.equals(cmd)) {						throw new MetaCodecException("Expect " + expect + ",but was " + cmd);					}				}			};		}		@Override		public Encoder getEncoder() {			//返回编码器return new Encoder() {				@Override				public IoBuffer encode(final Object message, final Session session) {                      //框架会在适当的时候调用编码器的encode()方法，前面说过如果响应的命令是DataCommand的时候假设不是zeroCopy的话，会出现问题。原因就在这里，因为如果不使用zeroCopy的话，返回给Gecko框架的是一个DataCommand的实例，这时候会调用到此方法，而此方法并没有按照DataCommand的格式进行编码，解码器会识别不了，所以容易出问题					return ((MetaEncodeCommand) message).encode();				}			};		}

前面还介绍到过MetaMorphosisBroker在启动时会注册请求类型与Processor的映射，见代码：

Java代码

private void registerProcessors() {
this.remotingServer.registerProcessor(GetCommand.class, new GetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));
this.remotingServer.registerProcessor(PutCommand.class, new PutProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));
this.remotingServer.registerProcessor(OffsetCommand.class, new OffsetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));
this.remotingServer.registerProcessor(HeartBeatRequestCommand.class, new VersionProcessor(this.brokerProcessor));
this.remotingServer.registerProcessor(QuitCommand.class, new QuitProcessor(this.brokerProcessor));
this.remotingServer.registerProcessor(StatsCommand.class, new StatsProcessor(this.brokerProcessor));
this.remotingServer.registerProcessor(TransactionCommand.class, new TransactionProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));
}

private void registerProcessors() {		this.remotingServer.registerProcessor(GetCommand.class, new GetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));		this.remotingServer.registerProcessor(PutCommand.class, new PutProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));		this.remotingServer.registerProcessor(OffsetCommand.class, new OffsetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));		this.remotingServer.registerProcessor(HeartBeatRequestCommand.class, new VersionProcessor(this.brokerProcessor));		this.remotingServer.registerProcessor(QuitCommand.class, new QuitProcessor(this.brokerProcessor));		this.remotingServer.registerProcessor(StatsCommand.class, new StatsProcessor(this.brokerProcessor));		this.remotingServer.registerProcessor(TransactionCommand.class, new TransactionProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));}

依据注册的类型，Gecko框架将会根据解析出来的命令实例调用处理器的不同方法，并返回不同请求的响应。下面让我们来看看不同的处理到底做了些什么事情？因为是Broker针对请求的处理，所以所有的Processor都在server工程中，先上类图：

所有的处理器均实现了RequestProcessor接口，该接口由Gecko框架定义，RequestProcessor类中只定义了两个方法：

Java代码

public interface RequestProcessor<T extends RequestCommand> {
/**
* 处理请求
*
* @param request请求命令
* @param conn 请求来源的连接
*/
public void handleRequest(T request, Connection conn);
/**
* 用户自定义的线程池，如果提供，那么请求的处理都将在该线程池内执行
*
* @return
*/
public ThreadPoolExecutor getExecutor();
}

public interface RequestProcessor<T extends RequestCommand> {    /**     * 处理请求     *      * @param request请求命令     * @param conn 请求来源的连接     */    public void handleRequest(T request, Connection conn);    /**     * 用户自定义的线程池，如果提供，那么请求的处理都将在该线程池内执行     *      * @return     */    public ThreadPoolExecutor getExecutor();}

所以，加上上一篇文章，我们可以得出MetaQ的大致网络处理流程图解如下：

http://soledede.com/

大家可以加我个人微信号：scccdgf

或者关注soledede的微信公众号：soledede

微信公众号：

分享到：

MetaQ技术内幕——源码分析(八) | MetaQ技术内幕——源码分析(六)

2014-05-31 14:03
浏览 2014
评论(0)
分类:编程语言
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论