MultiplexingChannelSelector的初始化过程:
public void configure(Context context) {
//获取Header的值
this.headerName = context.getString(CONFIG_MULTIPLEX_HEADER_NAME,
DEFAULT_MULTIPLEX_HEADER);
Map<String, Channel> channelNameMap = getChannelNameMap();
//获取默认的Channel
defaultChannels = getChannelListFromNames(
context.getString(CONFIG_DEFAULT_CHANNEL), channelNameMap);
//获取Mapping的值
Map<String, String> mapConfig =
context.getSubProperties(CONFIG_PREFIX_MAPPING);
//channelMapping变量存放了header变量中必须的Channel列表
channelMapping = new HashMap<String, List<Channel>>();
//将header对应的Channels存放到channelMapping变量中。
for (String headerValue : mapConfig.keySet()) {
List<Channel> configuredChannels = getChannelListFromNames(
mapConfig.get(headerValue),
channelNameMap);
//This should not go to default channel(s)
//because this seems to be a bad way to configure.
if (configuredChannels.size() == 0) {
throw new FlumeException("No channel configured for when "
+ "header value is: " + headerValue);
}
if (channelMapping.put(headerValue, configuredChannels) != null) {
throw new FlumeException("Selector channel configured twice");
}
}
//If no mapping is configured, it is ok.
//All events will go to the default channel(s).
Map<String, String> optionalChannelsMapping =
context.getSubProperties(CONFIG_PREFIX_OPTIONAL + ".");
//以下这一整段代码中是赛选出Header对应那些可选Channel列表。
optionalChannels = new HashMap<String, List<Channel>>();
for (String hdr : optionalChannelsMapping.keySet()) {
List<Channel> confChannels = getChannelListFromNames(
optionalChannelsMapping.get(hdr), channelNameMap);
if (confChannels.isEmpty()) {
confChannels = EMPTY_LIST;
}
//Remove channels from optional channels, which are already
//configured to be required channels.
List<Channel> reqdChannels = channelMapping.get(hdr);
//Check if there are required channels, else defaults to default channels
if(reqdChannels == null || reqdChannels.isEmpty()) {
//如果header对应的必选Channel列表为空,那么deault就作为它的必选Channel
reqdChannels = defaultChannels;
}
for (Channel c : reqdChannels) {
//如果header对应的Channel是必选的,那么就在可选的列表中删除。
if (confChannels.contains(c)) {
confChannels.remove(c);
}
}
if (optionalChannels.put(hdr, confChannels) != null) {
throw new FlumeException("Selector channel configured twice");
}
}
}
在看看MultiplexingChannelSelector中getRequiredChannels和getOptionalChannels方法,这两个方法也是根据HeaderName来获取Channel列表的: @Override
public List<Channel> getRequiredChannels(Event event) {
String headerValue = event.getHeaders().get(headerName);
//headerValue不存在,就获取默认
if (headerValue == null || headerValue.trim().length() == 0) {
return defaultChannels;
}
//根据headerName获取必选的Channel列表
List<Channel> channels = channelMapping.get(headerValue);
//This header value does not point to anything
//Return default channel(s) here.
//必选列表为null,则返回默认的Channel列表
if (channels == null) {
channels = defaultChannels;
}
return channels;
}
@Override
public List<Channel> getOptionalChannels(Event event) {
String hdr = event.getHeaders().get(headerName);
//根据HeaderValue获取可选的channel列表
List<Channel> channels = optionalChannels.get(hdr);
//可选列表为null,则返回空列表。
if(channels == null) {
channels = EMPTY_LIST;
}
return channels;
} 自定义 Channel Selector