Fix unsubscribe blocked when consumer is closing or has closed

[wuYin]

Motivation

For the present consumer, Close() and Unsubscribe() handled by the same eventloop goroutine.
The eventloop exited after Close(), then unsubscribe event wouldn't be selected and handled anymore, lead to block.

example:

func main() {
	client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:            "topic-1",
		SubscriptionName: "my-sub",
	})
	if err != nil {
		log.Fatal(err)
	}
	
	defer consumer.Unsubscribe() // unintentional
	defer consumer.Close()
}

Unsubscribe() blocked:

Modifications

Check consumer state before send unsubscribe event, if consumer is closing or has closed, just logging it

Verifying this change

• Make sure that the change passes the CI checks.

[freeznet]

LGTM

[wolfstudy]

LGTM +1