Fix the issue that GroupBy may not call 'unsubscribe'

Is this for making the group emission thread-safe with a concurrent unsubscribe?