Закрыть канал RabbitMQ, находясь в BasicAcks

Question

У меня есть пул каналов RabbitMQ под одним соединением. И я пытаюсь реализовать функцию Publisher Confirms. Я создал пул таким образом, чтобы он создавал новые каналы по мере увеличения спроса на каналы. Но теперь я должен справиться и с закрытием этих каналов.

Я планирую реализовать решение, как в RabbitMQ , используя список outstandingConfirms.

Я пришел к проблеме, что мне нужно как-то закрыть канал ( когда я превышаю мой «мягкий» порог объектов в пуле), когда последнее сообщение в канале равно ack-ed или nack-ed.

Как видно из кода ниже, sender Параметр на самом деле является самим каналом, я подумал, что мог бы использовать это напрямую, чтобы закрыть канал, если больше нет ожидающих сообщений. Но я столкнулся с тем, что каналы не должны использоваться более чем одним потоком одновременно. Этот же канал будет доступен в пуле и может быть выбран приложением для использования.

Это подписки на события:

protected void OnBasicAcks(object sender, BasicAckEventArgs e)
{
    //sender = channel object
    //todo close channel after all pending messages (n)acked
}

protected void OnBasicNacks(object sender, BasicNackEventArgs e)
{
    //sender = channel object
    //todo close channel after all pending messages (n)acked
}

Также здесь I может прочитать, что в обработчиках обратного вызова разрешены операции блокировки:

Начиная с версии 3.5.0, обработчики обратного вызова приложения могут вызывать операции блокировки (такие как IModel.QueueDeclare или IModel.BasicCancel)

Но относится ли это и к закрытию самого канала?

Короче, это моя проблема. Мои вопросы:

1. Могу ли я использовать канал, пропущенный через параметр sender напрямую, для выполнения таких операций, как IModel.Close()?
2. Я просто хочу предотвратить блокировку каналов или другие странные вещи. Какой-нибудь совет или лучшая практика для моей проблемы?
3. И что-то совершенно другое: могу ли я создать новый канал, пока я использую метод OnBasicAcks?

Answer 1

Если вы используете последний RabbitMQ.Client (я думаю, v5.1), все обратные вызовы вызываются в ThreadPool с использованием asyn c, так что вы, как правило, можете вызывать эти методы в обратные вызовы. Однако я бы не стал перехватывать поток обратного вызова для выполнения каких-либо длительных задач, поскольку они управляются для каждого канала клиентом RabbitMQ.

Итак, чтобы ответить конкретно.

1. Должно быть хорошо, глядя на исходный код, он не блокирует ничего, что помешало бы вызову close. Просто будьте осторожны со всем, что в вашем коде может привести к взаимоблокировке.
2. Лучше всего контролировать доступ к сеансу в вашем собственном коде, чтобы не вызывать один канал из нескольких потоков, ожидающих что-нибудь должно произойти в указанном c порядке.
3. Должно работать нормально, вы находитесь в потоке задач TPL, поэтому нет причин, по которым это не должно работать.

Если вы Чтобы узнать больше о том, как работает клиент, вы можете посмотреть на source , в частности на файл ModelBase.cs , и на асинхронный рабочий сервис, который он создает для обратных вызовов.