Java 类com.rabbitmq.client.BuiltinExchangeType 实例源码

public static void main(String[] argv) throws Exception {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost("localhost");
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();

  channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);

  String message = getMessage(argv);

  channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
  System.out.println(" [x] Sent '" + message + "'");

  channel.close();
  connection.close();
}

public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection conn = factory.newConnection();
    Channel channel = conn.createChannel();

    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);

    String msg = getMessage(argv);

    Date date = new Date();
    SimpleDateFormat sdf = new SimpleDateFormat("dd MMM yyyy @ HH:mm:ss");
    String sDate = sdf.format(date);

    String finalMsg = sDate + ": " + msg;

    channel.basicPublish(EXCHANGE_NAME, "", null, finalMsg.getBytes("UTF-8"));
    System.out.println("Emmited message: " + finalMsg);

    channel.close();
    conn.close();
}

public static void main(String[] argv) throws Exception {

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

    String severity = getSeverity(argv);
    String message = getMessage(argv);

    channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"));
    System.out.println(" [x] Sent '" + severity + "':'" + message + "'");

    channel.close();
    connection.close();
  }

private void initRabbitMQ() throws IOException {
    Server.LOGGER.info("Initialization of the Notifications channel");
    mRabbitMQManager.getChannel().addShutdownListener(new ShutdownListener() {

        @Override
        public void shutdownCompleted(ShutdownSignalException cause) {
            cause.printStackTrace();
        }
    });
    mRabbitMQManager.getChannel().exchangeDeclare("Pub", BuiltinExchangeType.FANOUT, true);
    String queueName = mRabbitMQManager.getChannel().queueDeclare().getQueue();
    mRabbitMQManager.getChannel().queueBind(queueName, "Pub", "");

    mRabbitMQManager.getChannel().basicConsume(queueName, true, new RabbitMQConsumer(this, mRabbitMQManager.getChannel()));
    Server.LOGGER.info("Initialization of the Pub channel done.");
}

public static void main(String[] argv) {
  Connection connection = null;
  Channel channel = null;
  try {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");

    connection = factory.newConnection();
    channel = connection.createChannel();

    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

    String routingKey = getRouting(argv);
    String message = getMessage(argv);

    channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
    System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");

  }
  catch  (Exception e) {
    e.printStackTrace();
  }
  finally {
    if (connection != null) {
      try {
        connection.close();
      }
      catch (Exception ignore) {}
    }
  }
}

private void recvTopicMessage() throws Throwable {

        String exchange = queue.getName();
        // String exchangeType = (String) queue.getProperties().get(RabbitMqVo.RABBITMQ_EXCHANGE_TYPE);
        // final boolean autoAck = (Boolean) queue.getProperties().get(RabbitMqVo.RABBITMQ_AUTOACK);

        String exchangeType = queue.getExchangeType();
        final boolean autoAck = queue.isAutoAck();

        // RabbitMqSource mqSource = (RabbitMqSource) MqContainer.getInstance().getMqSourceManager().getMqSource(queue.getMsKey());
        mqSource = (RabbitMqSource) MqContainer.getInstance().getMqSourceManager().getMqSource(queue.getMsKey());
        channel = mqSource.getChannel();
        channel.exchangeDeclare(exchange, exchangeType);
        String queueName = channel.queueDeclare().getQueue();

        if (BuiltinExchangeType.FANOUT.getType().equalsIgnoreCase(exchangeType)) {
            channel.queueBind(queueName, exchange, "");
        } else {
            if (null == binding) {
                throw new TangYuanException("exchange[" + exchange + "], the binding is empty.");
            }
            List<BindingPattern> patterns = binding.getPatterns();
            if (null == patterns || 0 == patterns.size()) {
                throw new TangYuanException("exchange[" + exchange + "], the binding is empty.");
            }
            for (BindingPattern bp : patterns) {
                channel.queueBind(queueName, exchange, bp.getPattern());
            }
        }
        // channel.basicQos(prefetchCount);// TODO
        running = true;
        // boolean asynReceiveMessages = (Boolean) queue.getProperties().get(RabbitMqVo.RABBITMQ_C_ASYNRECEIVEMESSAGES);
        boolean asynReceiveMessages = queue.isAsynReceive();
        recv(asynReceiveMessages, queueName, autoAck, null);
    }

private void sendTopicMessage(RabbitMqChannelVo queue, RoutingVo rVo, Object arg, boolean useTx, MqServiceContext context) throws Throwable {
    RabbitMqSource mqSource = (RabbitMqSource) MqContainer.getInstance().getMqSourceManager().getMqSource(queue.getMsKey());
    Channel channel = null;
    Throwable tx = null;
    String exchange = queue.getName();
    try {
        // String exchangeType = (String) queue.getProperties().get(RabbitMqVo.RABBITMQ_EXCHANGE_TYPE);
        String exchangeType = queue.getExchangeType();

        String routingKey = null;
        if (BuiltinExchangeType.FANOUT.getType().equalsIgnoreCase(exchangeType)) {
            routingKey = "";
        } else {
            routingKey = getExchangeRoutingKey(rVo, arg);
        }

        if (null == routingKey) {
            throw new TangYuanException("when sending a topic[" + exchange + "], the routingKey is empty, arg: " + arg);
        }

        channel = mqSource.getChannel();
        channel.exchangeDeclare(exchange, exchangeType);

        if (useTx) {
            channel.txSelect();
            context.addTransactionObject(new RabbitMqTransactionObject(mqSource, channel));
        }

        byte[] body = getBody(arg);
        channel.basicPublish(exchange, routingKey, null, body);

        // log.info("send message to topic[" + exchange + "]: " + ((XCO) arg).toXMLString());
        log.info("send message to exchange[" + exchange + "]: " + ((XCO) arg).toXMLString());
    } catch (Throwable e) {
        tx = e;
    } finally {
        if (null != channel && !useTx) {
            // closeChannel(channel);
            mqSource.closeChannel(channel);
        }
        if (null != tx) {
            throw tx;
        }
    }
}

public RabbitMQReceiver(String host, int port, String exchangeName, BuiltinExchangeType policy, ReceivingHandler receiver) throws Exception {
    super();
    this.host = host;
    this.port = port;
    this.queueName = null;
    this.exchangeName = exchangeName;
    this.topicbindingKey = null;
    this.policy = policy;
    this.receiver = receiver;

    init();
}

public RabbitMQReceiver(String host, int port, String exchangeName, String topicbindingKey, BuiltinExchangeType policy, ReceivingHandler receiver) throws Exception {
    super();
    this.host = host;
    this.port = port;
    this.queueName = null;
    this.exchangeName = exchangeName;
    this.topicbindingKey = topicbindingKey;
    this.policy = policy;
    this.receiver = receiver;

    init();
}

public RabbitMQExchanger(String host, int port, String exchangeName, BuiltinExchangeType policy) throws Exception {
    this.host = host;
    this.port = port;
    this.exchangeName = exchangeName;
    this.policy = policy;
    init();
}

public RabbitMQExchanger(String host, int port, String exchangeName, BuiltinExchangeType policy, RPCEventHandler eventHandler) throws Exception {
    this.host = host;
    this.port = port;
    this.exchangeName = exchangeName;
    this.policy = policy;
    this.eventHandler = eventHandler;
    init();
}

public void testExFanoutSending() throws Exception {

    RabbitMQExchanger exchanger = new RabbitMQExchanger("127.0.0.1", 5672, "order1", BuiltinExchangeType.FANOUT);
    for (int i=0; i<10; i++) {
        exchanger.send("hello: "+i, MessageProperties.MINIMAL_PERSISTENT_BASIC);
        Thread.sleep(100);
    }
    exchanger.close();
}

public void testExTopicSending() throws Exception {

    RabbitMQExchanger exchanger = new RabbitMQExchanger("127.0.0.1", 5672, "order", BuiltinExchangeType.TOPIC);
    for (int i=0; i<10; i++) {
        exchanger.send(("hello: "+i).getBytes(), ((i%2==0)?"topicA.hahah":"topicB.hahah"), MessageProperties.MINIMAL_PERSISTENT_BASIC);
        Thread.sleep(100);
    }
    exchanger.close();
}

@Override
public AMQP.Exchange.DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type)
    throws IOException {
  return channel.exchangeDeclare(exchange, type);
}

@Override
public AMQP.Exchange.DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type,
    boolean durable) throws IOException {
  return channel.exchangeDeclare(exchange, type, durable);
}

@Override
public AMQP.Exchange.DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type,
    boolean durable, boolean autoDelete, Map<String, Object> arguments) throws IOException {
  return channel.exchangeDeclare(exchange, type, durable, autoDelete, arguments);
}

@Override
public AMQP.Exchange.DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type,
    boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
    throws IOException {
  return channel.exchangeDeclare(exchange, type, durable, autoDelete, internal, arguments);
}

@Override
public void exchangeDeclareNoWait(String exchange, BuiltinExchangeType type, boolean durable,
    boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException {
  channel.exchangeDeclareNoWait(exchange, type, durable, autoDelete, internal, arguments);
}

private void initRabbitMQ() throws IOException {
    LOGGER.info("Initialization of the Notifications channel");
    mRabbitMQManager.getChannel().exchangeDeclare("Pub", BuiltinExchangeType.FANOUT, true);
    LOGGER.info("Initialization of the Pub channel done.");
}

public static void init(Connection connection) throws Exception {
    Channel channel = connection.createChannel();
    channel.exchangeDeclare(EXCHANGE_CHAT, BuiltinExchangeType.TOPIC, true);
    channel.close();
}

/**
 * declare a new exchange with no extra arguments
 *
 * @param exchangeName the name of the exchange
 * @param exchangeType the exchange type
 * @throws IOException if an error is encountered
 */
void declareExchange(String exchangeName, BuiltinExchangeType exchangeType) throws IOException;