// "guest"/"guest" by default, limited to localhost connections
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("name");
factory.setPassword("pw");
//factory.setUri("amqp://name:pw@localhost:5672");
Connection conn = factory.newConnection();
Channel channel = conn.createChannel();
// autoDelete(true) : exchange 를 사용하는 곳이 없으면 삭제. durable 속성과는 상관없음.
// durable : 브로커 재시작 후에도 큐의 상태를 유지 여부.
//channel.exchangeDeclare("aiif.test", "direct", false/*durable*/, false/*autoDelete*/, null);
// exclusive
// connection close 되면 큐 삭제 됨.
// durable, autoDelete 속성과는 상관없음.
//channel.queueDeclare("ExclusiveQueue", false/*durable*/, true/*exclusive*/, false/*autoDelete*/, null);
// auto delete
// queue 에 consuming 하는 client 가 없으면 삭제.
// 일단 queue 를 생성 후 consuming 하는 client 가 있다가 없어진 경우에 삭제됨.
channel.queueDeclare("AutoDeleteQueue", false, false, false, null);
//channel.queueBind("AutoDeleteQueue", "aiif.test", "AutoDeleteQueue");
//DefaultConsumer consumer = new DefaultConsumer(channel);
final boolean autoAck = true;
String tag = channel.basicConsume("AutoDeleteQueue", autoAck/*auto ack*/, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message;
if (body == null) {
message = "";
System.out.println("body null");
} else {
message = new String(body);
System.out.println("message : " + message);
}
Map<String, Object> h = properties.getHeaders();
System.out.println(h);
String routingKey = envelope.getRoutingKey();
String contentType = properties.getContentType();
long deliveryTag = envelope.getDeliveryTag();
System.out.println("routingKey : " + routingKey);
System.out.println("contentType : " + contentType);
System.out.println("deliveryTag : " + deliveryTag);
MQMessage mqMessage = MQMessageMaker.make(message);
System.out.println(mqMessage.toString());
MQMessageBodyHB mqMessageBodyHB = (MQMessageBodyHB)mqMessage.getBody();
System.out.println(mqMessageBodyHB.getLoad_level());
System.out.println(mqMessageBodyHB.getAiif_id());
if(!autoAck) {
channel.basicAck(deliveryTag, false/*multiple*/);
}
}
});
System.out.println("subscribe queue : AutoDeleteQueue");
//Thread.sleep(10000);
channel.basicCancel(tag);
System.out.println("unsubscribe queue : AutoDeleteQueue");
Thread.sleep(10000);
channel.queueDelete("AutoDeleteQueue");
System.out.println("delete queue : AutoDeleteQueue");
Thread.sleep(10000);
channel.close();
System.out.println("channel close");
Thread.sleep(10000);
conn.close();
System.out.println("connection close");