基本概念
过程如下:
主要是把交换机设置为直连的方式direct直连的方式然后把数据发送给交换机。
交换机再通过路由的Key值转发到队列上。
每一个客户端,都有一个队列,监听自己的队列就可以收到RabbitMQ转发过来的数据了。
代码与实例
其中recv1就绑定了error路由键
recv2绑定了info、error、warning路由键
对应的RabbitMQ如下:
程序运行截如下:
当只发送error作为路由键时:
此时发送端如下:
接收端如下:
当发送端key值为info时!
接收端如下:
代码如下:
package routing;
import com.飞狐通达信接口,rabbitmq.client.*;
import util.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Recv1 {
private static final String EXCHANGE_NAME = 'test_exchange_direct';
private static final String QUEUE_NAME = 'test_queue_direct_1';
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnect();
final Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicQos(1);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, 'error');
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, 'utf-8');
System.out.println('[1] Recv msg : ' + msg);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println('[1] done');
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
package routing;
import com.飞狐通达信接口,rabbitmq.client.*;
import util.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Recv2 {
private static final String EXCHANGE_NAME = 'test_exchange_direct' +
'';
private static final String QUEUE_NAME = 'test_queue_direct_2';
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnect();
final Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicQos(1);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, 'error');
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, 'info');
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, 'warning');
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, 'utf-8');
System.out.println('[2] Recv msg : ' + msg);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println('[2] done');
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
package routing;
import com.飞狐通达信接口,rabbitmq.client.Channel;
import com.飞狐通达信接口,rabbitmq.client.Connection;
import util.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Send {
private static final String EXCHANGE_NAME = 'test_exchange_direct';
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnect();
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME, 'direct');
String msg = 'Hello direct!';
String routingKey = 'info';
System.out.println('Send msg : ' + msg);
channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
channel.close();
connection.close();
}
}
源码打包下载地址:
文章为作者独立观点,不代表股票交易接口观点