process_queue.py 文件源码

def handle(self, *args, **options):
        if not options['queue_names']:
            raise CommandError('Queue names (--queues) not specified')

        queue_names = [queue_name.rstrip() for queue_name in options['queue_names'].split(',')]

        logger.debug('[django-eb-sqs] Connecting to SQS: {}'.format(', '.join(queue_names)))

        sqs = boto3.resource(
            'sqs',
            region_name=settings.AWS_REGION,
            config=Config(retries={'max_attempts': settings.AWS_MAX_RETRIES})
        )
        queues = [sqs.get_queue_by_name(QueueName=queue_name) for queue_name in queue_names]

        logger.debug('[django-eb-sqs] Connected to SQS: {}'.format(', '.join(queue_names)))

        worker = WorkerFactory.default().create()

        while True:
            for queue in queues:
                messages = queue.receive_messages(
                    MaxNumberOfMessages=settings.MAX_NUMBER_OF_MESSAGES,
                    WaitTimeSeconds=settings.WAIT_TIME_S,
                )

                for msg in messages:
                    self._process_message(msg, worker)