我有一个应用程序，它使用 RabbitMQ 从某个生产者那里接收消息，解析它们并将它们发送到其他地方。问题是我的应用程序没有完全使用 CPU 能力。它使用不超过 25% 的 CPU。从分析器看这个截图：
这是执行处理的最大部分的代码：
public class Consumer {
private static final String QUEUE_NAME = "MdnaMessagesQueue";
private static int i = 1;
private final ConnectionFactory factory;
private final boolean autoAck = false;
private final Connection connection;
private final Channel channel;
private String response;
private Sender sender;
private Logger LOG = Logger.getLogger(Consumer.class);
private ExecutorService es;
public Consumer() throws IOException{
es = Executors.newFixedThreadPool(8);
factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection(es);
channel = connection.createChannel();
sender = new Sender();
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.basicQos(25);
Properties props = new Properties();
try {
props.load(new FileInputStream("/home/mikhail/bzrrep/DLP/DLPServer/src/main/java/log4j.properties"));
} catch (Exception e){
LOG.error(e);
}
PropertyConfigurator.configure(props);
}
/**
* is used to receive messages from the queue
* @param customerId the id of current customer
* @throws IOException
* @throws InterruptedException
*/
public void receive(final String customerId) throws IOException,InterruptedException{
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
final QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(QUEUE_NAME, autoAck, "myConsumerTag",
new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException{
BasicProperties replyProperties = new BasicProperties.Builder().correlationId(properties.getCorrelationId()).build();
long deliveryTag = envelope.getDeliveryTag();
try{
LOG.info(" [" +(i++) +"] Received from first queue ");
byte[] dataArr = body;
MimeParser mimeParser = new MimeParser(true);
Filter filter = new Filter();
ByteArrayInputStream bais1 = new ByteArrayInputStream(dataArr);
MessageInfo mi = mimeParser.parseMessages(bais1);
//checking for compliance with rules
boolean messageAccepted = filter.getMessageAcceptability(mi.getPlainText(), customerId);
response = filter.getResult();
if(messageAccepted){
//sending to the other queue
sender.send(dataArr);
channel.basicPublish("", properties.getReplyTo(), replyProperties, response.getBytes());//need to add responce
} else {
channel.basicPublish("", properties.getReplyTo(), replyProperties, response.getBytes());
}
} catch (Exception e){
LOG.error(e);
}finally {
channel.basicAck(deliveryTag, false);
}
}
});
}
}
这是分析器的快照：
如何解决这个问题呢？