com.rabbitmq.client.Channel类的使用及代码示例

x33g5p2x  于2022-01-18 转载在 其他  
字(12.4k)|赞(0)|评价(0)|浏览(206)

本文整理了Java中com.rabbitmq.client.Channel类的一些代码示例，展示了Channel类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Channel类的具体详情如下：
包路径：com.rabbitmq.client.Channel
类名称：Channel

Channel介绍

暂无

代码示例

代码示例来源：origin: testcontainers/testcontainers-java

@Test
public void simpleRabbitMqTest() throws IOException, TimeoutException {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost(rabbitMq.getContainerIpAddress());
  factory.setPort(rabbitMq.getMappedPort(RABBITMQ_PORT));
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();
  channel.exchangeDeclare(RABBIQMQ_TEST_EXCHANGE, "direct", true);
  String queueName = channel.queueDeclare().getQueue();
  channel.queueBind(queueName, RABBIQMQ_TEST_EXCHANGE, RABBITMQ_TEST_ROUTING_KEY);
  // Set up a consumer on the queue
  final boolean[] messageWasReceived = new boolean[1];
  channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
      messageWasReceived[0] = Arrays.equals(body, RABBITMQ_TEST_MESSAGE.getBytes());
    }
  });
  // post a message
  channel.basicPublish(RABBIQMQ_TEST_EXCHANGE, RABBITMQ_TEST_ROUTING_KEY, null, RABBITMQ_TEST_MESSAGE.getBytes());
  // check the message was received
  assertTrue("The message was received", Unreliables.retryUntilSuccess(5, TimeUnit.SECONDS, () -> {
    if (!messageWasReceived[0]) {
      throw new IllegalStateException("Message not received yet");
    }
    return true;
  }));
}

代码示例来源：origin: Graylog2/graylog2-server

public void stop() throws IOException {
  if (channel != null && channel.isOpen()) {
    try {
      channel.close();
    } catch (TimeoutException e) {
      LOG.error("Timeout when closing AMQP channel", e);
      channel.abort();
    }
  }
  if (connection != null && connection.isOpen()) {
    connection.close();
  }
}

代码示例来源：origin: apache/nifi

public void acknowledge(final GetResponse response) throws IOException {
  if (autoAcknowledge) {
    return;
  }
  getChannel().basicAck(response.getEnvelope().getDeliveryTag(), true);
}

代码示例来源：origin: apache/flink

@Override
protected void acknowledgeSessionIDs(List<Long> sessionIds) {
  try {
    for (long id : sessionIds) {
      channel.basicAck(id, false);
    }
    channel.txCommit();
  } catch (IOException e) {
    throw new RuntimeException("Messages could not be acknowledged during checkpoint creation.", e);
  }
}

代码示例来源：origin: apache/nifi

@Override
public void close() throws TimeoutException, IOException {
  if (closed) {
    return;
  }
  if (logger.isDebugEnabled()) {
    logger.debug("Closing AMQP channel for " + this.channel.getConnection().toString());
  }
  this.channel.close();
  closed = true;
}

代码示例来源：origin: vector4wang/spring-boot-quick

public static void main(String[] args) {
    try {
      ConnectionFactory factory = new ConnectionFactory();
      factory.setUsername("guest");
      factory.setPassword("guest");
      factory.setHost("60.205.191.82");
      factory.setPort(5672);
      Connection conn = factory.newConnection();
      Channel channel = conn.createChannel();

//            channel.qu
      channel.queueDeclare("hello", false, false, false, null);
      String message = "Hello World!";
      channel.basicPublish("", "hello", null, message.getBytes());
      System.out.println(" [x] Sent '" + message + "'");

      channel.close();
      conn.close();
    } catch (IOException e) {
      e.printStackTrace();
    } catch (TimeoutException e) {
      e.printStackTrace();
    }
  }
}

代码示例来源：origin: Graylog2/graylog2-server

public void connect() throws IOException {
  final ConnectionFactory factory = new ConnectionFactory();
  factory.setHost(hostname);
  factory.setPort(port);
  factory.setVirtualHost(virtualHost);
  factory.setRequestedHeartbeat(heartbeatTimeout);
  channel = connection.createChannel();
    channel.basicQos(prefetchCount);
  connection.addShutdownListener(cause -> {
    if (cause.isInitiatedByApplication()) {
      LOG.info("Shutting down AMPQ consumer.");

代码示例来源：origin: zendesk/maxwell

public RabbitmqProducer(MaxwellContext context) {
  super(context);
  exchangeName = context.getConfig().rabbitmqExchange;
  props = context.getConfig().rabbitmqMessagePersistent ? MessageProperties.MINIMAL_PERSISTENT_BASIC : null;
  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost(context.getConfig().rabbitmqHost);
  factory.setPort(context.getConfig().rabbitmqPort);
  factory.setUsername(context.getConfig().rabbitmqUser);
  factory.setPassword(context.getConfig().rabbitmqPass);
  factory.setVirtualHost(context.getConfig().rabbitmqVirtualHost);
  try {
    this.channel = factory.newConnection().createChannel();
    if(context.getConfig().rabbitmqDeclareExchange) {
      this.channel.exchangeDeclare(exchangeName, context.getConfig().rabbitmqExchangeType, context.getConfig().rabbitMqExchangeDurable, context.getConfig().rabbitMqExchangeAutoDelete, null);
    }
  } catch (IOException | TimeoutException e) {
    throw new RuntimeException(e);
  }
}

代码示例来源：origin: zstackio/zstack

tracker = new MessageTracker();
ConnectionFactory connFactory = new ConnectionFactory();
List<Address> addresses = CollectionUtils.transformToList(serverIps, new Function<Address, String>() {
  @Override
connFactory.setAutomaticRecoveryEnabled(true);
connFactory.setRequestedHeartbeat(CloudBusGlobalProperty.RABBITMQ_HEART_BEAT_TIMEOUT);
connFactory.setNetworkRecoveryInterval((int) TimeUnit.SECONDS.toMillis(CloudBusGlobalProperty.RABBITMQ_NETWORK_RECOVER_INTERVAL));
connFactory.setConnectionTimeout((int) TimeUnit.SECONDS.toMillis(CloudBusGlobalProperty.RABBITMQ_CONNECTION_TIMEOUT));
  conn = connFactory.newConnection(addresses.toArray(new Address[]{}));
  logger.debug(String.format("rabbitmq connection is established on %s", conn.getAddress()));
  outboundQueue = new BusQueue(makeMessageQueueName(SERVICE_ID), BusExchange.P2P);
  Channel chan = channelPool.acquire();
  chan.queueDeclare(outboundQueue.getName(), false, false, true, queueArguments());
  chan.basicConsume(outboundQueue.getName(), true, consumer);
  chan.queueBind(outboundQueue.getName(), outboundQueue.getBusExchange().toString(), outboundQueue.getBindingKey());
  channelPool.returnChannel(chan);
  maid.construct();

代码示例来源：origin: yacy/yacy_grid_mcp

private void init() throws IOException {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setAutomaticRecoveryEnabled(true);
  factory.setHost(this.server);
  if (this.port > 0) factory.setPort(this.port);
  if (this.username != null && this.username.length() > 0) factory.setUsername(this.username);
  if (this.password != null && this.password.length() > 0) factory.setPassword(this.password);
  try {
    this.connection = factory.newConnection();
    //Map<String, Object> map = this.connection.getServerProperties();
    if (!this.connection.isOpen()) throw new IOException("no connection");
    this.channel = connection.createChannel();
    if (!this.channel.isOpen()) throw new IOException("no channel");
    this.queues = new ConcurrentHashMap<>();
  } catch (TimeoutException e) {
    throw new IOException(e.getMessage());
  }
}

代码示例来源：origin: spring-projects/spring-integration

private void testNackOrRequeue(boolean requeue) throws IOException, TimeoutException {
  Channel channel = mock(Channel.class);
  willReturn(true).given(channel).isOpen();
  Envelope envelope = new Envelope(123L, false, "ex", "rk");
  BasicProperties props = new BasicProperties.Builder().build();
  GetResponse getResponse = new GetResponse(envelope, props, "bar".getBytes(), 0);
  willReturn(getResponse).given(channel).basicGet("foo", false);
  Connection connection = mock(Connection.class);
  willReturn(true).given(connection).isOpen();
  willReturn(channel).given(connection).createChannel();
  ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
  willReturn(connection).given(connectionFactory).newConnection((ExecutorService) isNull(), anyString());
  CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
  AmqpMessageSource source = new AmqpMessageSource(ccf, "foo");
  Message<?> received = source.receive();
  verify(connection).createChannel();
  StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
      .acknowledge(requeue ? Status.REQUEUE : Status.REJECT);
  verify(channel).basicReject(123L, requeue);
  verify(connection).createChannel();
  ccf.destroy();
  verify(channel).close();
  verify(connection).close(30000);
}

代码示例来源：origin: berndruecker/flowing-retail-old

public void connect() throws IOException, TimeoutException {
 ConnectionFactory factory = new ConnectionFactory();
 factory.setHost("localhost");
 connection = factory.newConnection();
 channel = connection.createChannel();
 channel.exchangeDeclare(RabbitMqConsumer.EXCHANGE_NAME, "fanout", true); // publish/subscribe model
 System.out.println("Connected to RabbitMQ");
}

代码示例来源：origin: wmr513/reactive

public static void main(String[] args) throws Exception {
    Channel channel = AMQPCommon.connect();
    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicQos(1);
    channel.basicConsume("trade.eq.q", false, consumer);

    while (true) {
      QueueingConsumer.Delivery msg = consumer.nextDelivery();
      System.out.println("message received: " + new String(msg.getBody()));
      Thread.sleep(2000);
      channel.basicAck(msg.getEnvelope().getDeliveryTag(), false);
    }            
  }    
}

代码示例来源：origin: NationalSecurityAgency/lemongrenade

public GetDeadLetterMessages(String queueName) throws Exception {
  this.queueName = queueName;
  factory = new ConnectionFactory();
  factory.setHost(rabbitmq_host);
  connection = factory.newConnection();
  channel = connection.createChannel();
  //Add Dead letter queue
  channel.exchangeDeclare(queueName, "fanout", true);
  channel.queueDeclare(queueName, true, false, false, null);
  channel.queueBind(queueName, queueName, "");
  consumer = new QueueingConsumer(channel);
}

代码示例来源：origin: apache/apex-malhar

public void setup() throws IOException
{
 logger.debug("setting up receiver..");
 connFactory.setHost(host);
 connection = connFactory.newConnection();
 channel = connection.createChannel();
 channel.exchangeDeclare(exchange, "fanout");
 queueName = channel.queueDeclare().getQueue();
 channel.queueBind(queueName, exchange, "");
 tracingConsumer = new TracingConsumer(channel);
 cTag = channel.basicConsume(queueName, true, tracingConsumer);
}

代码示例来源：origin: imalexyang/ExamStack

@Bean
QueueingConsumer queueingConsumer() throws IOException {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost(messageQueueHostname);
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();
  channel.queueDeclare(Constants.ANSWERSHEET_DATA_QUEUE, true, false, false, null);
  QueueingConsumer consumer = new QueueingConsumer(channel);
  channel.basicConsume(Constants.ANSWERSHEET_DATA_QUEUE, true, consumer);
  return consumer;
}

代码示例来源：origin: spring-projects/spring-amqp

@Test
public void test(ConnectionFactory connectionFactory) throws Exception {
  Connection conn = connectionFactory.newConnection();
  Channel channel = conn.createChannel();
  DeclareOk declareOk = channel.queueDeclarePassive("rabbitAvailableTests.queue");
  assertEquals(0, declareOk.getConsumerCount());
  channel.close();
  conn.close();
}

代码示例来源：origin: zstackio/zstack

void init() {
  try {
    ConnectionFactory connFactory = new ConnectionFactory();
    List<Address> addresses = CollectionUtils.transformToList(bus.getServerIps(), new Function<Address, String>() {
      @Override
      public Address call(String arg) {
        return Address.parseAddress(arg);
      }
    });
    conn = connFactory.newConnection(addresses.toArray(new Address[]{}));
    chan = conn.createChannel();
    String name = MessageTracker.class.getName();
    chan.queueDeclare(name, true, false, true, null);
    chan.basicConsume(name, true, this);
    chan.queueBind(name, BusExchange.P2P.toString(), "#");
    chan.queueBind(name, BusExchange.BROADCAST.toString(), "#");
  } catch (Exception e) {
    throw new CloudRuntimeException(e);
  }
}

代码示例来源：origin: org.geoserver.community/gs-notification

public void receive(ReceiverService service) throws Exception {
  // let's setup evrything and start listening
  this.service = service;
  ConnectionFactory factory = createConnectionFactory();
  factory.setSaslConfig(new CustomSaslConfig());
  connection = factory.newConnection();
  channel = connection.createChannel();
  channel.exchangeDeclare("testExchange", "fanout");
  channel.queueDeclare(QUEUE_NAME, false, true, false, null);
  channel.queueBind(QUEUE_NAME, "testExchange", "testRouting");
  channel.basicConsume(QUEUE_NAME, true, newConsumer(channel));
}

代码示例来源：origin: gudaoxuri/dew

public void subscribeWithTopic(String topic, String routingKey, String queueName, Consumer<String> consumer) {
  Channel channel = rabbitAdapter.getConnection().createChannel(false);
  try {
    channel.queueDeclare(queueName, true, false, false, null);
    channel.exchangeDeclare(topic, BuiltinExchangeType.TOPIC, true);
    channel.queueBind(queueName, topic, routingKey);
    channel.basicQos(1);
    channel.basicConsume(queueName, false, getDefaultConsumer(channel, topic, consumer));
  } catch (IOException e) {
    logger.error("[MQ] Rabbit subscribeWithTopic error.", e);
  }
}

内容来源于网络，如有侵权，请联系作者删除！

Channel

相关文章

    查看更多
    微信公众号

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Channel类方法

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com