[2015-06-22 13:14:28,780] [ERROR] [sysName] [subSys or component] [Thread-9] [com.messagebus.client.handler.common.CommonLoopHandler] -*- stacktrace -*- : com.rabbitmq.client.ShutdownSignalException: clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=OK, class-id=0, method-id=0)
at com.rabbitmq.client.QueueingConsumer.handle(QueueingConsumer.java:203)
at com.rabbitmq.client.QueueingConsumer.nextDelivery(QueueingConsumer.java:220)
at com.messagebus.client.handler.common.CommonLoopHandler.handle(CommonLoopHandler.java:34)
at com.messagebus.client.handler.consume.ConsumerDispatchHandler.handle(ConsumerDispatchHandler.java:17)
at com.messagebus.client.handler.MessageCarryHandlerChain.handle(MessageCarryHandlerChain.java:72)
at com.messagebus.client.handler.consume.RealConsumer.handle(RealConsumer.java:44)
at com.messagebus.client.handler.MessageCarryHandlerChain.handle(MessageCarryHandlerChain.java:72)
at com.messagebus.client.handler.consume.ConsumerTagGenerator.handle(ConsumerTagGenerator.java:22)
at com.messagebus.client.handler.MessageCarryHandlerChain.handle(MessageCarryHandlerChain.java:72)
at com.messagebus.client.handler.consume.ConsumePermission.handle(ConsumePermission.java:37)
at com.messagebus.client.handler.MessageCarryHandlerChain.handle(MessageCarryHandlerChain.java:72)
at com.messagebus.client.handler.consume.ConsumeParamValidator.handle(ConsumeParamValidator.java:17)
at com.messagebus.client.handler.MessageCarryHandlerChain.handle(MessageCarryHandlerChain.java:72)
at com.messagebus.client.carry.GenericConsumer.run(GenericConsumer.java:50)
at java.lang.Thread.run(Thread.java:744)
Caused by: com.rabbitmq.client.ShutdownSignalException: clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=OK, class-id=0, method-id=0)
当然你也可以对日志内容进行特殊处理,让一条日志的所有内容以一行输出,但这样需要对日志框架进行定制,有时这并不受你控制。因此这里最好的选择是定制日志收集器。
源码问题定位
我们先来了解一下Flume源码中LineDeserializer的核心实现:
private String readLine() throws IOException {
StringBuilder sb = new StringBuilder();
int c;
int readChars = 0;
while ((c = in.readChar()) != -1) {
readChars++;
// FIXME: support \r\n
if (c == '\n') {
break;
}
sb.append((char)c);
if (readChars >= maxLineLength) {
logger.warn("Line length exceeds max ({}), truncating line!",
maxLineLength);
break;
}
}
if (readChars > 0) {
return sb.toString();
} else {
return null;
}
}
首先,构建一个StringBuilder,然后以字符为单位挨个读取,如果读取到换行符“\n”,则表示读取本条日志结束,跳出循环;否则将该字符串追加到StringBuilder中。与此同时会给读取的字符个数计数:如果读取的字符个数大于预先配置的一行日志的最大字符串长度,也会跳出循环。