Comments 6
Вот всегда мне была интересна эта наша, скажем там, чисто-ITшная "особенность"... Ну есть же "спека", ну возьми и сделай все "по черчежу". Но - нет... "треугольные колеса" :-)
простой и прозрачный пул каналов;
Хм... а что этот "пул" должен решать? Каналы же (если речь о каналах amqp) не шарятся. Вы не можете, грубо говоря, сделать consume в одном канале, а cancel сделать в другом. Какой смысл в этом "пуле"?
Ну и с точки зрения прикладной логики, кланал - это *сессия*. Она начинается с открытия канала, и заканачивается его закрытием. Мигрировать на другой канал она ("просто так") не может.
Такой ключ можно использовать сразу и для консумеров и паблишеров. Как было сказано выше, каналы между ними не должны пересекаться для повышения потокобезопасности.
То, что каналы для консумеров и паблишеров "не должны пересекаться" - это, мягко говоря, бред. Каким образом - при таком "непересечении" - вы собераетесь обеспечивать, например, транзакционность?!
При подключении создаётся TCP-подключение и служебный канал. Последний нужен для создания exchange, очередей и биндинга очереди к exchange.
За такое, в приличном обществе, я извиняюсь, "бьют канделябрами". Каким образом, при таком подходе, вы собираетесь обеспечивать, например, экслюзивность деклараций?
Дальше просто уже не смог читать :-(
Ребят... ну реально, ну вы сначала хотяб разберитесь с "как оно работает". Ну "спеки" все открытые - это не какой-то там секрет. И "спеки" нулевой серии amqp - это именно, что *спецификации* - в инженерном смысле... а не в нашем... ITшном, так сказать. Там же даже придумывать ничего не надо. Все описано/расписано - бери и делай.
Хм... а что этот "пул" должен решать? Каналы же (если речь о каналах amqp) не шарятся. Вы не можете, грубо говоря, сделать consume в одном канале, а cancel сделать в другом. Какой смысл в этом "пуле"?
Смысл, чтобы отслеживать созданные каналы и если попросят тот же самый, например, для паблишера в тот же самый exchange, в который уже отправляли сообщение, не создавать канал заново, а переиспользовать тот же самый. Так же он позволяет пройтись по всем каналам при закрытии соединения с RabbitMQ и аккуратно закрыть сначала каналы, а потом и подключение.
То, что каналы для консумеров и паблишеров "не должны пересекаться" - это, мягко говоря, бред. Каким образом - при таком "непересечении" - вы собераетесь обеспечивать, например, транзакционность?!
Простите, но причём тут транзакционность? Отправка и получение сообщение это одна операция, а не десять последовательных шагов. И без разницы, на каком канале я это делаю.
Да, консумер и паблишер могут прекрасно работать на одном канале. Ничего не сломается, наверное. Но лучше же их наверно разнести?
За такое, в приличном обществе, я извиняюсь, "бьют канделябрами". Каким образом, при таком подходе, вы собираетесь обеспечивать, например, экслюзивность деклараций?
Вы точно знакомы с RabbitMQ? Ниже кусок из документации на github.com/streadway/amqp
ExchangeDeclare declares an exchange on the server. If the exchange does notalready exist, the server will create it. If the exchange exists, the server verifies that it is of the provided type, durability and auto-delete flags.
Смысл, чтобы отслеживать созданные каналы и если попросят тот же самый, например, для паблишера в тот же самый exchange, в который уже отправляли сообщение, не создавать канал заново, а переиспользовать тот же самый.
Сдается мне, вы совсем не понимаете, что такое каналы в amqp :-) При чем тут, вообще exchange?! Еще раз... что со стороны сервера, что со стороны клиента - канал - это сессия. Более того... до 0.10 для сервера, конкретный канал - это не просто сессия... это ещё и "привязка" конкретного экземпляра клиента... со всеми его qos, таймингами, фреймингами и т.п.
А если же вы "шарите" канал по прикладной логике (что само по себе не сильно хорошо... ибо не понятно, как recovery делать, например... но не суть) - то, имхо, сильно проще делать это явно... не привязываясь к каким-то exchange... а то можно ненароком подумать, что паблишер не может в разные exchange сообщения посылать. Или, что ему для этого разные каналы нужны :-)
Так же он позволяет пройтись по всем каналам при закрытии соединения с RabbitMQ и аккуратно закрыть сначала каналы, а потом и подключение.
?! Оно, насколько я помню, всю жизнь работало так: посылаешь серверу connection.close, он тебе в ответ засылает пачку channel.close, ты их обрабатываешь, в конце концов получая от сервера connection.close-ok... после чего закрываешь сокет.
В любом случае, хранить каналы в отрыве от родительского подключения - это такое. Ну не умеет "девятка" в нормальный детач. "Десятка" - умеет. Но "кролики" - не умеют в "десятку" :-)
Простите, но причём тут транзакционность? Отправка и получение сообщение это одна операция, а не десять последовательных шагов. И без разницы, на каком канале я это делаю.
Если прием "запроса" (грубо, basic.ack, как ответ на basic.delivery), и отправка ответа (грубо, basic.publish) происходят в разных каналах, то вы не сможете связать эти две команды через tx.commit. Они обе в одном канале с tx.commit должны быть, который должен быть в том же канале, из которого получен basic.delivery. А это - в свою очередь - будет тот канал в котором был basic.consume.
Так понятнее?
Да, консумер и паблишер могут прекрасно работать на одном канале. Ничего не сломается, наверное. Но лучше же их наверно разнести?
Вот как раз с точностью до наоброт. Если этого одна "логика" - например, какой-нибудь сервис типа запрос/ответ - то прием запроса, и отправка ответа могут конечно "жить" в разных каналах. Но это, скажем так, сильно сужает использование возможностей amqp.
Вы точно знакомы с RabbitMQ?
А причем тут exchange?! У него просто нет экслюзивности. В отличие, например, от queue... а их, как я понял, вы тож через свой "служебный канал" объявляете. Т.е. - если на пальцах - очереди, внезапно, могут быть и "приватными"... консумить которые можно только через тот же канал, через который они были объявлены. И это не просто так придумано :-)
Мне кажется, проблема в том, что я явно не подсветил в начале статьи - я не делаю ещё одну реализацию протокола amqp. Я использую готовую библиотеку, где всё уже сделано за меня. Но там нет автореконнекта.
Сдается мне, вы совсем не понимаете, что такое каналы в amqp :-) При чем тут, вообще exchange?! Еще раз... что со стороны сервера, что со стороны клиента - канал - это сессия. Более того... до 0.10 для сервера, конкретный канал - это не просто сессия... это ещё и "привязка" конкретного экземпляра клиента... со всеми его qos, таймингами, фреймингами и т.п.
Да, канал это сессия. Под exchange в моём ответе имелось ввиду его имя, чтобы по имени в пуле на один и тот же exchange вернуть один и тот же канал. Т.е. какдый раз, когда будут отправлять сообщение в один и тот же exchange, будет использоваться одна и таже сессия. Причём у меня не только коннект описан, но и паблишер, который получит строго один раз сессию и будет её использовать до получения ошибки или до его удаления. При получении ошибки канал/сессия будут пересозданы и опять будут единственные для этого паблишера.
Также и с косьюмером, он строго один раз получает сессию по имени очереди и не меняет её до ошибки или удаления консьюмера.
Я никак не подменяю логику работы с каналами в github.com/streadway/amqp. Сделана просто обёртка над ней для реконнекта. И пока косьюмер/паблишер жив, он использует один и тот же канал, который он получит при запуске.
а то можно ненароком подумать, что паблишер не может в разные exchange сообщения посылать. Или, что ему для этого разные каналы нужны :-)
Как это возможно в моей реализации? Здесь указывается конкретный exchange для паблишера при его создании.
?! Оно, насколько я помню, всю жизнь работало так: посылаешь серверу connection.close, он тебе в ответ засылает пачку channel.close, ты их обрабатываешь, в конце концов получая от сервера connection.close-ok... после чего закрываешь сокет.
Повторю, я не выдумываю свою реализацию взаимодействия по протоколу amqp, используется готовая библиотека github.com/streadway/amqp, я сделал только реконнект поверх неё.
Вы расписываете про то, как на уровне протокола работает, я это знаю. Но я не пишу свою реализацию, зачем, если есть хорошая библиотека.
Закрытие каналов в github.com/streadway/amqp это не только отправить сообщения по протоколу, это ещё и грохнуть созданные структуры. И для прозрачности сделано по примеру github.com/rabbitmq/rabbitmq-tutorials/blob/master/go/send.go, создали подключение, потом канал, при завершении выполняем в обратном порядке.
Вот как раз с точностью до наоброт. Если этого одна "логика" - например, какой-нибудь сервис типа запрос/ответ - то прием запроса, и отправка ответа могут конечно "жить" в разных каналах. Но это, скажем так, сильно сужает использование возможностей amqp.
Поясните, в чём сужение? Напомню, я не делаю свою реализацию протокола.
А причем тут exchange?! У него просто нет экслюзивности. В отличие, например, от queue... а их, как я понял, вы тож через свой "служебный канал" объявляете. Т.е. - если на пальцах - очереди, внезапно, могут быть и "приватными"... консумить которые можно только через тот же канал, через который они были объявлены. И это не просто так придумано :-)
Exclusive (used by only one connection and the queue will be deleted when that connection closes)
An exclusive queue can only be used (consumed from, purged, deleted, etc) by its declaring connection
Из документации на RabbitMQ, где здесь речь про канал? А соединение у меня одно.
Как это возможно в моей реализации? Здесь указывается конкретный exchange для паблишера при его создании.
Поясните, в чём сужение? Напомню, я не делаю свою реализацию протокола.
Ну вот в том и сужение возможностей, что протокол позволяет слать сообщения в разные exchange в рамках одной сессии, а вы не позволяете.
Ну так у меня сам паблишер ограничен только конкретным exchange, ему он передаётся при создании. Был бы в моей реализации паблишер, который шлёт в разные exchange, тогда да, это было серьёзное сужение, т.к. в текущей реализации я бы городил отдельные сессии. Я подумаю в эту сторону.
Сейчас можно выдернуть сессию через GetChannelFromPool и послать через неё в другой Exchange.
Автореконнекты к RabbitMQ в Go