title: Celery/Django/Redis same task getting executed multiple times
body: <p>In my current project, what I need to do is to get data from 700+ endpoints and then send that data to another 700+ endpoints. My approach is to use Django Celery Redis, Put 70+ endpoints on each worker so that there are approx 10 workers which will retrieve the data and then post the data.</p>
  <p>For this, I am using Chord to do the parallel task and then calculate the time it takes.</p>
  <p>The problem is that Celery is running the same task multiple times. <code>task_get_data</code> is the main method that first gets the list of websites then splits it into groups of 70 each and then calls <code>task_post_data</code> using Chord.</p>
  <p>In the output below you can see <code>website_A</code>, <code>website_B</code> etc. multiple times, I have manually checked my data and everything and there is no repetition of websites but when the celery task is submitted, multiple entries are created.</p>
  <p>Also, Is there any way to monitor the number of workers and what are they processing?</p>
  <p>Below is the code</p>
  <pre><code>os.environ.setdefault('DJANGO_SETTINGS_MODULE','django_backend.settings')
  
  app = Celery('django_backend', backend='redis://localhost:6379', broker='redis://localhost:6379')
  
  app.config_from_object('django.conf:settings', namespace='CELERY')
  # app.config_from_object('django.conf:settings')
  app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
  
  @app.task(bind=True)
  def debug_task(self):
      print('Request: {0!r}'.format(self.request))
  
  def post_data(json_obj, website):
      for items in json_obj:
          md = md + items['data']
      n = 50
      list_of_objects = [md[i:i+n] for i in range(0, len(md), n)]
      
      print(&quot;Starting to post data using post__data&quot;)
      for items in list_of_objects:
          data_details = PostDataDetails()
          data_details.data = items
          post_data_response = &quot;&quot;
          try:
              post_data_response = client.post_data(post_data_details = data_details)
              
              print(&quot;Successfully posted data with status code &quot; + str(post_data_response.status) + &quot; for &quot; + website)
          except Exception as e:
             
      return post_data_response.status
  
  def read_from_file():
          with open('./django_backend/data.json') as json_file:
              data = json.load(json_file)
              # print(data)
          return data
  
  def split_list(l,n):
      website_list = [l[i:i+n] for i in range(0, len(l), n)]
      return website_list
  
  
  #-----------this is the main method---------
  @app.task(bind=True, name=&quot;task_get_data&quot;)
  def task_get_data(self):
      start_time = datetime.datetime.now()
      try:
          website = task_helper.get_data()
          task_helper.write_logs(&quot;info&quot;, &quot;Successfully read from cache&quot;)
      except Exception as e:
          task_helper.write_logs(&quot;error&quot;, &quot;Error in reading from cache. Error&quot; + str(e))
    
      website_list_chunks = split_list(list(website.keys()),70)
      callback = get_time_difference.s(start_time)
  
      try:
          task_helper.write_logs(&quot;info&quot;, &quot;Starting the task_post_poller_data task to post  data&quot;)
          header = [task_post_data.s(website_list) for website_list in website_list_chunks]
          result = chord(header)(callback)
          print(result)
          task_helper.write_logs(&quot;info&quot;, &quot;Successfully completed the task to post  data&quot;)
      except Exception as e:
          task_helper.write_logs(&quot;error&quot;, &quot;Error in creating task_post_data task. Error&quot; + str(e))
  
  
  @app.task(bind=True, name=&quot;task_post_data&quot;)
  def task_post_data(self,website_list=None) -&gt; dict:
      json_object_response = True
      post_data_response = None
      for website in website_list:
          if json_object_response:
              file_data = read_from_file()
              try:
                  post_data_response = post_data(file_data, website)
                  # pass
              except Exception as e:
                  print(&quot;error&quot;, &quot;Error in creating task_post_poller_data task. Error&quot; + str(e))
      return post_data_response    
  </code></pre>
  <p>I am running the code using
  <code>celery -A django_backend worker -l debug --purge</code> command and I am submitting the task using this command</p>
  <pre><code>python manage.py shell
  &gt;&gt;&gt;from django_backend.tasks import task_get_data
  &gt;&gt;&gt;task_get_data.delay()
  </code></pre>
  <p>Below is the output from console</p>
  <pre><code>[2021-07-20 19:54:54,789: INFO/ForkPoolWorker-3] Successfully posted  data with status code 200 for  website_D
  [2021-07-20 19:54:54,835: INFO/ForkPoolWorker-5] Successfully posted  data with status code 200 for  website_E
  [2021-07-20 19:54:54,840: INFO/ForkPoolWorker-2] Successfully posted  data with status code 200 for  website_B
  [2021-07-20 19:54:54,843: INFO/ForkPoolWorker-1] Successfully posted  data with status code 200 for  website_A
  [2021-07-20 19:54:54,882: INFO/ForkPoolWorker-6] Successfully posted  data with status code 200 for  website_P
  [2021-07-20 19:54:54,891: INFO/ForkPoolWorker-8] Successfully posted  data with status code 200 for  website_I
  [2021-07-20 19:54:54,895: INFO/ForkPoolWorker-4] Successfully posted  data with status code 200 for  website_R
  [2021-07-20 19:54:55,021: INFO/ForkPoolWorker-3] Successfully posted  data with status code 200 for  website_D
  [2021-07-20 19:54:55,025: INFO/ForkPoolWorker-7] Successfully posted  data with status code 200 for  website_C
  [2021-07-20 19:54:55,073: INFO/ForkPoolWorker-2] Successfully posted  data with status code 200 for  website_B
  [2021-07-20 19:54:55,086: INFO/ForkPoolWorker-1] Successfully posted  data with status code 200 for  website_A
  </code></pre>