声明一个主题交换机队列，并且将队列和交换机进行绑定，同时监听这个队列
        :param exchange_name:
        :param routing_key:
        :param queue_name:
        :param callback:
        :return:
350039
def declare_consuming(self, exchange_name, routing_key, queue_name, callback):
        
        self.bind_topic_exchange(exchange_name, routing_key, queue_name)
        self.declare_basic_consuming(
            queue_name=queue_name,
            callback=callback
        )
229799
def serv_forever(self, func):
        
        # 如果任务队列不为空则声明队列
        if self.task_queue:
            self.ch.queue_declare(
                queue=self.task_queue, durable=self.durable)
        if self.store_queue:
            self.ch.queue_declare(
                queue=self.store_queue, durable=self.durable)

        def callback(ch, method, properties, body):
            """
            amqp的回调函数
            :param ch: 链道
            :param method: 默认的方法
            :param properties: 默认
            :param body: 接收到的数据任务
            :return:
            """
            if self.prefetch_count > 1:
                # 如果任务队列的并发执行数大于1，则启动线程
                # 否则，单线程
                task = _RabbitmqTask(func, (self, body), self.no_ack, ch, method, properties)
                task.start()
            else:
                func(self, body)
                # 任务结束后，确认任务
                if not self.no_ack:
                    self.ch.basic_ack(delivery_tag=self.method.delivery_tag)

        # 声明队列
        self.ch.basic_qos(prefetch_count=self.prefetch_count)
        self.ch.basic_consume(callback, queue=self.task_queue, no_ack=self.no_ack)
        # 启动消费者
        self.ch.start_consuming()
941308
def _get_queue_name(cls, queue_name=None):
        
        if queue_name is None and cls.queue_name is None:
            raise LimpydJobsException("Queue's name not defined")
        if queue_name is None:
            return cls.queue_name
        return queue_name
344633
AsteriskQueueImpl getQueueByName(String queueName)
    {
        refreshQueueIfForced(queueName);

        AsteriskQueueImpl queue = getInternalQueueByName(queueName);
        if (queue == null)
        {
            logger.error("Requested queue '" + queueName + "' not found!");
        }

        return queue;
    }
872011
def get_queue(name, options = {})
      response = action("GetQueueUrl", {"QueueName" => Queue.sanitize_name("#{queue_prefix}#{name}")}.merge(options))
      
      if response.success?
        Queue.new(self, response.doc.xpath("//QueueUrl").text)
      end

    rescue Bisques::AwsActionError => e
      raise unless e.code == "AWS.SimpleQueueService.NonExistentQueue"
    end
559937
def _encoded_routing_key(self):
        """"""
        topic = self.topic
        if config.conf["topic_prefix"]:
            topic = ".".join((config.conf["topic_prefix"].rstrip("."), topic))
        return topic.encode("utf-8")
933730
def etf_swap_in(self, etf_name, amount, _async=False):
        

        params = {}
        path = '/etf/swap/in'
        params['etf_name'] = etf_name
        params['amount'] = amount

        return api_key_post(params, path,  _async=_async)
345470
def unbind(exchange, opts = {})
      check_queue_deleted!

      if exchange.respond_to?(:remove_route)

        # we can do the unbinding ourselves
        exchange.remove_route opts.fetch(:routing_key, @name), self
      else

        # we need the channel to lookup the exchange
        @channel.queue_unbind self, opts.fetch(:routing_key, @name), exchange
      end
    end
109275
public static function createFromNamesWithoutRoutingKey(string $amqpExchangeName, string $amqpQueueName) : self
    {
        return new self(AmqpExchange::createDirectExchange($amqpExchangeName), AmqpQueue::createWith($amqpQueueName), null);
    }
906412
def _generate_queues(queues, exchange, platform_queue):
        
        return set([
            Queue('celery', exchange, routing_key='celery'),
            Queue(platform_queue, exchange, routing_key='#'),
        ] + [
            Queue(q_name, exchange, routing_key=q_name)
            for q_name in queues
        ])
1094096
public function ask($routing_key, $message, $timeout = 10, $exchange = null)
    {
        return $this->amqp->ask($exchange ?: $this->exchange, $routing_key, $message, $timeout);
    }
85095
public AmqpTemplate getAmqpTemplate(String queueName)
    {
        // このメソッドを呼び出す前に、
        // this#initContextMap()を利用してcontextMapを初期しておくこと。 
        if (this.contextMap == null)
        {
            return null;
        }

        RabbitmqClusterContext context = this.contextMap.get(queueName);
        if (context == null)
        {
            return null;
        }

        return context.getAmqpTemplate();
    }
1168854
public function unbindExchange($destinationName, $sourceName, $routingKey = '')
    {
        // WARNING: only amqp 1.6.0 and above supports exchange unbinding
        if (phpversion('amqp') <= '1.4.0') {
            return;
        }

        $exchange = $this->produceExchangeInstance($destinationName);

        $exchange->unbind($sourceName, $routingKey);
    }
882624
public static QueueName getQueueName(String appName, String flowName, KeyValue keyValue) {
    return getQueueName(appName, flowName, keyValue.getBuffer(), keyValue.getRowOffset(), keyValue.getRowLength());
  }
613899
def unbind_redundant_bindings(queue, routing_keys)
      return unless http_api_use_enabled?

      bindings.each do |dest, keys|
        next unless dest == queue.name
        keys.reject { |key| routing_keys.include?(key) }.each do |key|
          logger.debug "removing redundant binding #{queue.name} <--> #{key}"
          queue.unbind(exchange, routing_key: key)
        end
      end
    end
25938
def get_etf_records(self, etf_name, offset, limit, _async=False):
        
        params = {}
        path = '/etf/list'
        params['etf_name'] = etf_name
        params['offset'] = offset
        params['limit'] = limit

        return api_key_get(params, path, _async=_async)
351608
public function send($routing_key, $message, $exchange = null, $type = Amqp::TYPE_TOPIC)
    {
        $this->amqp->send($exchange ?: $this->exchange, $routing_key, $message, $type);
    }