code: def create(self, ensure_keys_exist=True, mkstream=False):
          
          if ensure_keys_exist:
              for key in self.keys:
                  if not self.database.exists(key):
                      msg_id = self.database.xadd(key, {'': ''}, id=b'0-1')
                      self.database.xdel(key, msg_id)
                  elif self.database.type(key) != b'stream':
                      raise ValueError('Consumer group key "%s" exists and is '
                                       'not a stream. To prevent data-loss '
                                       'this key will not be deleted.')
  
          resp = {}
  
          # Mapping of key -> last-read message ID.
          for key, value in self.keys.items():
              try:
                  resp[key] = self.database.xgroup_create(key, self.name, value,
                                                          mkstream)
              except ResponseError as exc:
                  if exception_message(exc).startswith('BUSYGROUP'):
                      resp[key] = False
                  else:
                      raise
          return resp
docs: Create the consumer group and register it with the group's stream keys.
  
          :param ensure_keys_exist: Ensure that the streams exist before creating
              the consumer group. Streams that do not exist will be created.
          :param mkstream: Use the "MKSTREAM" option to ensure stream exists (may
              require unstable version of Redis).