diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 9cb3442..d0c36fe 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -39,7 +39,7 @@
 	lookupService internal.LookupService
 	auth          auth.Provider
 
-	handlers            map[internal.Closable]bool
+	handlers            internal.ClientHandlers
 	producerIDGenerator uint64
 	consumerIDGenerator uint64
 }
@@ -86,14 +86,14 @@
 	}
 	c.rpcClient = internal.NewRPCClient(url, c.cnxPool)
 	c.lookupService = internal.NewLookupService(c.rpcClient, url)
-	c.handlers = make(map[internal.Closable]bool)
+	c.handlers = internal.NewClientHandlers()
 	return c, nil
 }
 
 func (c *client) CreateProducer(options ProducerOptions) (Producer, error) {
 	producer, err := newProducer(c, &options)
 	if err == nil {
-		c.handlers[producer] = true
+		c.handlers.Add(producer)
 	}
 	return producer, err
 }
@@ -103,7 +103,7 @@
 	if err != nil {
 		return nil, err
 	}
-	c.handlers[consumer] = true
+	c.handlers.Add(consumer)
 	return consumer, nil
 }
 
@@ -145,9 +145,7 @@
 }
 
 func (c *client) Close() {
-	for handler := range c.handlers {
-		handler.Close()
-	}
+	c.handlers.Close()
 }
 
 func (c *client) namespaceTopics(namespace string) ([]string, error) {
diff --git a/pulsar/internal/client_handlers.go b/pulsar/internal/client_handlers.go
new file mode 100644
index 0000000..1ecfdd9
--- /dev/null
+++ b/pulsar/internal/client_handlers.go
@@ -0,0 +1,52 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package internal
+
+import "sync"
+
+// ClientHandlerMap is a simple concurrent-safe map for the client type
+type ClientHandlers struct {
+	handlers map[Closable]bool
+	l        *sync.RWMutex
+}
+
+func NewClientHandlers() ClientHandlers {
+	return ClientHandlers{
+		handlers: map[Closable]bool{},
+		l:        &sync.RWMutex{},
+	}
+}
+func (h *ClientHandlers) Add(c Closable) {
+	h.l.Lock()
+	defer h.l.Unlock()
+	h.handlers[c] = true
+}
+func (h *ClientHandlers) Val(c Closable) bool {
+	h.l.RLock()
+	defer h.l.RUnlock()
+	return h.handlers[c]
+}
+
+func (h *ClientHandlers) Close() {
+	h.l.Lock()
+	defer h.l.Unlock()
+
+	for handler := range h.handlers {
+		handler.Close()
+	}
+}
diff --git a/pulsar/internal/client_handlers_test.go b/pulsar/internal/client_handlers_test.go
new file mode 100644
index 0000000..baaf899
--- /dev/null
+++ b/pulsar/internal/client_handlers_test.go
@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package internal
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClientHandlers(t *testing.T) {
+	h := NewClientHandlers()
+	assert.NotNil(t, h.l)
+	assert.Equal(t, h.handlers, map[Closable]bool{})
+
+	closable := &testClosable{false}
+	h.Add(closable)
+	assert.True(t, h.Val(closable))
+
+	h.Close()
+	t.Log("closable is: ", closable.closed)
+	assert.True(t, closable.closed)
+}
+
+type testClosable struct {
+	closed bool
+}
+
+func (t *testClosable) Close() {
+	t.closed = true
+}