Я не вижу amqp_consumer.py или amqp_producer.py в тарболе, поэтому воспроизвести ошибку сложно.
RabbitMQ завершает соединения, выпуская их неподтвержденные сообщения для повторной доставки другим клиентам, когда операционная система сообщает, что сокет закрыт. Ваши симптомы очень странные, потому что даже kill -9 должно привести к правильной очистке сокета TCP.
Некоторые люди заметили проблемы с сокетами, которые дольше сохраняются, чем при работе с межсетевым экраном или устройством NAT между клиентами AMQP и сервером. Может ли это быть проблемой здесь, или вы запускаете все на локальном хосте? Кроме того, в какой операционной системе вы используете различные компоненты системы?
ETA: Из вашего комментария ниже я предполагаю, что, пока вы работаете на сервере в Linux, вы можете запускать клиенты в Windows. Если это так, то может быть так, что драйвер Windows TCP неправильно закрывает сокеты, что отличается от поведения kill-9 в Unix. (В Unix ядро должным образом закроет TCP-соединения для любого прерванного процесса.)
Если это так, то плохая новость заключается в том, что RabbitMQ может освобождать ресурсы только при закрытом сокете, поэтому, если клиентская операционная система не делает этого, она ничего не может сделать. Это то же самое, что почти любой другой сервис на основе TCP.
Хорошая новость 1019 *, однако, заключается в том, что AMQP поддерживает опцию «пульса» именно в этих случаях, когда сетевая структура не заслуживает доверия. Вы можете попробовать включить сердцебиение. Когда они включены, если сервер не получает трафик в течение настраиваемого интервала, он решает, что соединение должно быть разорвано.
Плохая новость 1023 *, однако, заключается в том, что я не думаю, что py-amqplib в настоящее время поддерживает сердцебиение. Хотя стоит попробовать!