diff --git a/socketio/server.py b/socketio/server.py
index 56152f5..0dd3979 100644
--- a/socketio/server.py
+++ b/socketio/server.py
@@ -30,6 +30,9 @@ class Server(object):
                  packets. Custom json modules must have ``dumps`` and ``loads``
                  functions that are compatible with the standard library
                  versions.
+    :param async_handlers: If set to ``True``, event handlers are executed in
+                           separate threads. To run handlers synchronously,
+                           set to ``False``. The default is ``False``.
     :param kwargs: Connection parameters for the underlying Engine.IO server.
 
     The Engine.IO configuration supports the following settings:
@@ -62,7 +65,7 @@ class Server(object):
                             ``False``.
     """
     def __init__(self, client_manager=None, logger=False, binary=False,
-                 json=None, **kwargs):
+                 json=None, async_handlers=False, **kwargs):
         engineio_options = kwargs
         engineio_logger = engineio_options.pop('engineio_logger', None)
         if engineio_logger is not None:
@@ -70,6 +73,7 @@ class Server(object):
         if json is not None:
             packet.Packet.json = json
             engineio_options['json'] = json
+        engineio_options['async_handlers'] = False
         self.eio = engineio.Server(**engineio_options)
         self.eio.on('connect', self._handle_eio_connect)
         self.eio.on('message', self._handle_eio_message)
@@ -99,6 +103,8 @@ class Server(object):
         self.manager = client_manager
         self.manager_initialized = False
 
+        self.async_handlers = async_handlers
+
         self.async_mode = self.eio.async_mode
 
     def on(self, event, handler=None, namespace=None):
@@ -412,7 +418,14 @@ class Server(object):
         namespace = namespace or '/'
         self.logger.info('received event "%s" from %s [%s]', data[0], sid,
                          namespace)
-        r = self._trigger_event(data[0], namespace, sid, *data[1:])
+        if self.async_handlers:
+            self.start_background_task(self._handle_event_internal, self, sid,
+                                       data, namespace, id)
+        else:
+            self._handle_event_internal(self, sid, data, namespace, id)
+
+    def _handle_event_internal(self, server, sid, data, namespace, id):
+        r = server._trigger_event(data[0], namespace, sid, *data[1:])
         if id is not None:
             # send ACK packet with the response returned by the handler
             # tuples are expanded as multiple arguments
@@ -426,10 +439,10 @@ class Server(object):
                 binary = False  # pragma: nocover
             else:
                 binary = None
-            self._send_packet(sid, packet.Packet(packet.ACK,
-                                                 namespace=namespace,
-                                                 id=id, data=data,
-                                                 binary=binary))
+            server._send_packet(sid, packet.Packet(packet.ACK,
+                                                   namespace=namespace,
+                                                   id=id, data=data,
+                                                   binary=binary))
 
     def _handle_ack(self, sid, namespace, id, data):
         """Handle ACK packets from the client."""
diff --git a/tests/test_server.py b/tests/test_server.py
index fafacf7..bcc2638 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -21,13 +21,15 @@ class TestServer(unittest.TestCase):
 
     def test_create(self, eio):
         mgr = mock.MagicMock()
-        s = server.Server(client_manager=mgr, binary=True, foo='bar')
+        s = server.Server(client_manager=mgr, binary=True,
+                          async_handlers=True, foo='bar')
         s.handle_request({}, None)
         s.handle_request({}, None)
-        eio.assert_called_once_with(**{'foo': 'bar'})
+        eio.assert_called_once_with(**{'foo': 'bar', 'async_handlers': False})
         self.assertEqual(s.manager, mgr)
         self.assertEqual(s.eio.on.call_count, 3)
         self.assertEqual(s.binary, True)
+        self.assertEqual(s.async_handlers, True)
         self.assertEqual(mgr.initialize.call_count, 1)
 
     def test_on_event(self, eio):
@@ -428,6 +430,8 @@ class TestServer(unittest.TestCase):
         self.assertRaises(ValueError, s.register_namespace, 123)
         self.assertRaises(ValueError, s.register_namespace, Dummy)
         self.assertRaises(ValueError, s.register_namespace, Dummy())
+        self.assertRaises(ValueError, s.register_namespace,
+                          namespace.Namespace)
 
     def test_logger(self, eio):
         s = server.Server(logger=False)
@@ -444,7 +448,8 @@ class TestServer(unittest.TestCase):
 
     def test_engineio_logger(self, eio):
         server.Server(engineio_logger='foo')
-        eio.assert_called_once_with(**{'logger': 'foo'})
+        eio.assert_called_once_with(**{'logger': 'foo',
+                                       'async_handlers': False})
 
     def test_custom_json(self, eio):
         # Warning: this test cannot run in parallel with other tests, as it
@@ -460,7 +465,8 @@ class TestServer(unittest.TestCase):
                 return '+++ decoded +++'
 
         server.Server(json=CustomJSON)
-        eio.assert_called_once_with(**{'json': CustomJSON})
+        eio.assert_called_once_with(**{'json': CustomJSON,
+                                       'async_handlers': False})
 
         pkt = packet.Packet(packet_type=packet.EVENT,
                             data={six.text_type('foo'): six.text_type('bar')})
@@ -471,6 +477,13 @@ class TestServer(unittest.TestCase):
         # restore the default JSON module
         packet.Packet.json = json
 
+    def test_async_handlers(self, eio):
+        s = server.Server(async_handlers=True)
+        s._handle_eio_message('123', '2["my message","a","b","c"]')
+        s.eio.start_background_task.assert_called_once_with(
+            s._handle_event_internal, s, '123', ['my message', 'a', 'b', 'c'],
+            '/', None)
+
     def test_start_background_task(self, eio):
         s = server.Server()
         s.start_background_task('foo', 'bar', baz='baz')