Spring Boot中通过RabbitTemplate主动pull(get)消息的例子

import java.io.IOException;
import java.util.Properties;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class RabbitAdminService {

private static final Logger LOGGER = LoggerFactory.getLogger(RabbitAdminService.class);

@Autowired
AmqpAdmin rabbitAdmin;
@Autowired
RabbitTemplate rabbitTemplate;

public int getCount(String queueName) {
Properties properties = rabbitAdmin.getQueueProperties(queueName);
return (Integer)properties.get(RabbitAdmin.QUEUE_MESSAGE_COUNT);
}

@PostConstruct
public void init(){
new Thread( new Runnable(){
@Override
public void run(){
while(true){
try{
Thread.sleep(1000);
processQueue(MQConstants.INDUSTRYSMS_QUEUE);
}catch(Exception e){

}
}

}
}).start();
}
public static class ChannelCallbackImpl implements ChannelCallback<Message>{
private String queueName;
public ChannelCallbackImpl(String queueName){
this.queueName = queueName;
}
@Override
public Message doInRabbit(Channel channel) {
GetResponse response = null;
try {
response = channel.basicGet(queueName, false);
 Message message = Message.parseFrom(response.getBody());

if(message != null){
int mobiles = message.getMessageTosList().size();
LOGGER.info("号码：{}", mobiles);
}
channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
return message;
}
catch(Exception e) {
try {
channel.basicNack(response.getEnvelope().getDeliveryTag(), false, true);
} catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
}
return null;
}
}
public  void processQueue(String queueName) {
rabbitTemplate.execute(new ChannelCallbackImpl(queueName));
}
}