Close this consumer thread and its underlying connection.

def close(self):
        """Close this consumer thread and its underlying connection.
        """
        try:
            if self.consumer:
                self.requeue_messages(m for _, m in iter_queue(self.delay_queue))
                self.consumer.close()
        except ConnectionError:
            pass