diff --git a/lib/mysql2/em.rb b/lib/mysql2/em.rb
index 329b3080e..15cf10d5d 100644
--- a/lib/mysql2/em.rb
+++ b/lib/mysql2/em.rb
@@ -3,6 +3,8 @@
 
 module Mysql2
   module EM
+    class ReadTimeout < ::RuntimeError; end
+
     class Client < ::Mysql2::Client
       module Watcher
         def initialize(client, deferable)
@@ -41,6 +43,9 @@ def query(sql, opts = {})
         if ::EM.reactor_running?
           super(sql, opts.merge(async: true))
           deferable = ::EM::DefaultDeferrable.new
+          if @read_timeout
+            deferable.timeout(@read_timeout, Mysql2::EM::ReadTimeout.new)
+          end
           @watch = ::EM.watch(socket, Watcher, self, deferable)
           @watch.notify_readable = true
           deferable
diff --git a/spec/em/em_spec.rb b/spec/em/em_spec.rb
index c57d474dd..8338f7d79 100644
--- a/spec/em/em_spec.rb
+++ b/spec/em/em_spec.rb
@@ -65,6 +65,24 @@
       end.to raise_error('some error')
     end
 
+    it "should timeout if we wait longer than :read_timeout" do
+      errors = []
+      EM.run do
+        client = Mysql2::EM::Client.new DatabaseCredentials['root'].merge(read_timeout: 1)
+        defer = client.query "SELECT sleep(2)"
+        defer.callback do
+          # This _shouldn't_ be run, but it needed to prevent the specs from
+          # freezing if this test fails.
+          EM.stop_event_loop
+        end
+        defer.errback do |err|
+          errors << err
+          EM.stop_event_loop
+        end
+      end
+      expect(errors).to eq([Mysql2::EM::ReadTimeout.new])
+    end
+
     context 'when an exception is raised by the client' do
       let(:client) { Mysql2::EM::Client.new DatabaseCredentials['root'] }
       let(:error) { StandardError.new('some error') }