diff --git a/opentelemetry-api/src/opentelemetry/util/_time.py b/opentelemetry-api/src/opentelemetry/util/_time.py
index ceaca22e8d4..a3fd113ce3c 100644
--- a/opentelemetry-api/src/opentelemetry/util/_time.py
+++ b/opentelemetry-api/src/opentelemetry/util/_time.py
@@ -28,7 +28,6 @@
     def _time_ns() -> int:
         return int(time() * 1e9)
 
-
 else:
     from time import time_ns
 
diff --git a/opentelemetry-sdk/setup.cfg b/opentelemetry-sdk/setup.cfg
index e78448dd820..eb6b81d5e15 100644
--- a/opentelemetry-sdk/setup.cfg
+++ b/opentelemetry-sdk/setup.cfg
@@ -45,6 +45,8 @@ include_package_data = True
 install_requires =
     opentelemetry-api == 1.7.1
     opentelemetry-semantic-conventions == 0.26b1
+    # FIXME Remove when 3.6 is no longer supported
+    dataclasses == 0.8; python_version < '3.7'
 
 [options.packages.find]
 where = src
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/__init__.py
index 1f306f68b31..5b003647aac 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/__init__.py
@@ -12,17 +12,28 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# pylint: disable=function-redefined,too-many-ancestors
-
-from abc import ABC, abstractmethod
 from atexit import register, unregister
 from logging import getLogger
-from typing import Optional
+from threading import Lock
+from typing import Optional, Sequence
 
 from opentelemetry._metrics import Meter as APIMeter
 from opentelemetry._metrics import MeterProvider as APIMeterProvider
 from opentelemetry._metrics import _DefaultMeter
+from opentelemetry._metrics.instrument import Counter as APICounter
+from opentelemetry._metrics.instrument import Histogram as APIHistogram
+from opentelemetry._metrics.instrument import (
+    ObservableCounter as APIObservableCounter,
+)
 from opentelemetry._metrics.instrument import (
+    ObservableGauge as APIObservableGauge,
+)
+from opentelemetry._metrics.instrument import (
+    ObservableUpDownCounter as APIObservableUpDownCounter,
+)
+from opentelemetry._metrics.instrument import Synchronous
+from opentelemetry._metrics.instrument import UpDownCounter as APIUpDownCounter
+from opentelemetry.sdk._metrics.instrument import (
     Counter,
     Histogram,
     ObservableCounter,
@@ -30,6 +41,9 @@
     ObservableUpDownCounter,
     UpDownCounter,
 )
+from opentelemetry.sdk._metrics.metric_exporter import MetricExporter
+from opentelemetry.sdk._metrics.metric_reader import MetricReader
+from opentelemetry.sdk._metrics.view import View
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.util.instrumentation import InstrumentationInfo
 
@@ -37,6 +51,8 @@
 
 
 class Meter(APIMeter):
+    """See `opentelemetry._metrics.Meter`."""
+
     def __init__(
         self,
         instrumentation_info: InstrumentationInfo,
@@ -46,37 +62,51 @@ def __init__(
         self._instrumentation_info = instrumentation_info
         self._meter_provider = meter_provider
 
-    def create_counter(self, name, unit=None, description=None) -> Counter:
-        # FIXME implement this method
-        pass
+    def create_counter(self, name, unit=None, description=None) -> APICounter:
+        # pylint: disable=protected-access
+        return self._meter_provider._create_counter(
+            self._instrumentation_info, name, unit, description
+        )
 
     def create_up_down_counter(
         self, name, unit=None, description=None
-    ) -> UpDownCounter:
-        # FIXME implement this method
-        pass
+    ) -> APIUpDownCounter:
+        # pylint: disable=protected-access
+        return self._meter_provider._create_up_down_counter(
+            self._instrumentation_info, name, unit, description
+        )
 
     def create_observable_counter(
         self, name, callback, unit=None, description=None
-    ) -> ObservableCounter:
-        # FIXME implement this method
-        pass
+    ) -> APIObservableCounter:
+        # pylint: disable=protected-access
+        return self._meter_provider._create_observable_counter(
+            self._instrumentation_info, name, unit, description
+        )
 
-    def create_histogram(self, name, unit=None, description=None) -> Histogram:
-        # FIXME implement this method
-        pass
+    def create_histogram(
+        self, name, unit=None, description=None
+    ) -> APIHistogram:
+        # pylint: disable=protected-access
+        return self._meter_provider._create_histogram(
+            self._instrumentation_info, name, unit, description
+        )
 
     def create_observable_gauge(
         self, name, callback, unit=None, description=None
-    ) -> ObservableGauge:
-        # FIXME implement this method
-        pass
+    ) -> APIObservableGauge:
+        # pylint: disable=protected-access
+        return self._meter_provider._create_observable_gauge(
+            self._instrumentation_info, name, unit, description
+        )
 
     def create_observable_up_down_counter(
         self, name, callback, unit=None, description=None
-    ) -> ObservableUpDownCounter:
-        # FIXME implement this method
-        pass
+    ) -> APIObservableUpDownCounter:
+        # pylint: disable=protected-access
+        return self._meter_provider._create_observable_up_down_counter(
+            self._instrumentation_info, name, unit, description
+        )
 
 
 class MeterProvider(APIMeterProvider):
@@ -84,57 +114,127 @@ class MeterProvider(APIMeterProvider):
 
     def __init__(
         self,
+        metric_exporters: Sequence[MetricExporter] = (),
+        metric_readers: Sequence[MetricReader] = (),
+        views: Sequence[View] = (),
         resource: Resource = Resource.create({}),
         shutdown_on_exit: bool = True,
+        use_always_matching_view: bool = True,
     ):
-        self._resource = resource
+        self._lock = Lock()
         self._atexit_handler = None
+        self.__use_always_matching_view = use_always_matching_view
 
         if shutdown_on_exit:
             self._atexit_handler = register(self.shutdown)
 
-        self._metric_readers = []
-        self._metric_exporters = []
-        self._views = []
-        self._shutdown = False
+        self.__metric_readers = metric_readers
 
-    def get_meter(
-        self,
-        name: str,
-        version: Optional[str] = None,
-        schema_url: Optional[str] = None,
-    ) -> Meter:
+        for metric_reader in self._metric_readers:
+            metric_reader._register_meter_provider(self)
 
-        if self._shutdown:
-            _logger.warning(
-                "A shutdown `MeterProvider` can not provide a `Meter`"
-            )
-            return _DefaultMeter(name, version=version, schema_url=schema_url)
+        self.__metric_exporters = metric_exporters
 
-        return Meter(InstrumentationInfo(name, version, schema_url), self)
+        self.__views = views
 
-    def shutdown(self):
-        # FIXME implement a timeout
+        if self.__use_always_matching_view:
+            self.__views = [*self.__views, View(instrument_name="*")]
 
-        if self._shutdown:
-            _logger.warning("shutdown can only be called once")
-            return False
+        self.__resource = resource
+        self._shutdown = False
+        self.__synchronous_instruments = []
+        self.__asynchronous_instruments = []
 
-        result = True
+    @property
+    def _metric_readers(self):
+        return self.__metric_readers
 
-        for metric_reader in self._metric_readers:
-            result = result and metric_reader.shutdown()
+    @property
+    def _use_always_matching_view(self):
+        return self.__use_always_matching_view
 
-        for metric_exporter in self._metric_exporters:
-            result = result and metric_exporter.shutdown()
+    @property
+    def _resource(self):
+        return self.__resource
 
-        self._shutdown = True
+    @property
+    def _metric_exporters(self):
+        return self.__metric_exporters
 
-        if self._atexit_handler is not None:
-            unregister(self._atexit_handler)
-            self._atexit_handler = None
+    @property
+    def _views(self):
+        return self.__views
 
-        return result
+    @property
+    def _synchronous_instruments(self):
+        return self.__synchronous_instruments
+
+    @property
+    def _asynchronous_instruments(self):
+        return self.__asynchronous_instruments
+
+    def _create_instrument(
+        self, instrument_type, instrumentation_info, name, unit, description
+    ):
+
+        instrument = instrument_type(
+            instrumentation_info, name, unit=unit, description=description
+        )
+
+        with self._lock:
+            if isinstance(instrument, Synchronous):
+                self.__synchronous_instruments.append(instrument)
+
+            else:
+                self.__asynchronous_instruments.append(instrument)
+
+        return instrument
+
+    def _create_counter(
+        self, instrumentation_info, name, unit, description
+    ) -> APICounter:
+        return self._create_instrument(
+            Counter, instrumentation_info, name, unit, description
+        )
+
+    def _create_up_down_counter(
+        self, instrumentation_info, name, unit, description
+    ) -> APIUpDownCounter:
+        return self._create_instrument(
+            UpDownCounter, instrumentation_info, name, unit, description
+        )
+
+    def _create_observable_counter(
+        self, instrumentation_info, name, unit, description
+    ) -> APIObservableCounter:
+        return self._create_instrument(
+            ObservableCounter, instrumentation_info, name, unit, description
+        )
+
+    def _create_histogram(
+        self, instrumentation_info, name, unit, description
+    ) -> APIHistogram:
+        return self._create_instrument(
+            Histogram, instrumentation_info, name, unit, description
+        )
+
+    def _create_observable_gauge(
+        self, instrumentation_info, name, unit, description
+    ) -> ObservableGauge:
+        return self._create_instrument(
+            ObservableGauge, instrumentation_info, name, unit, description
+        )
+
+    def _create_observable_up_down_counter(
+        self, instrumentation_info, name, unit, description
+    ) -> ObservableUpDownCounter:
+        return self._create_instrument(
+            ObservableUpDownCounter,
+            instrumentation_info,
+            name,
+            unit,
+            description,
+        )
 
     def force_flush(self) -> bool:
 
@@ -161,56 +261,40 @@ def force_flush(self) -> bool:
 
         return metric_reader_result and metric_exporter_result
 
-    def register_metric_reader(self, metric_reader: "MetricReader") -> None:
-        # FIXME protect this method against race conditions
-        self._metric_readers.append(metric_reader)
-
-    def register_metric_exporter(
-        self, metric_exporter: "MetricExporter"
-    ) -> None:
-        # FIXME protect this method against race conditions
-        self._metric_exporters.append(metric_exporter)
-
-    def register_view(self, view: "View") -> None:
-        # FIXME protect this method against race conditions
-        self._views.append(view)
-
-
-class MetricReader(ABC):
-    def __init__(self):
-        self._shutdown = False
-
-    @abstractmethod
-    def collect(self):
-        pass
-
     def shutdown(self):
-        # FIXME this will need a Once wrapper
-        self._shutdown = True
+        # FIXME implement a timeout
+
+        if self._shutdown:
+            _logger.warning("shutdown can only be called once")
+            return False
 
+        result = True
 
-class MetricExporter(ABC):
-    def __init__(self):
-        self._shutdown = False
+        for metric_reader in self._metric_readers:
+            result = result and metric_reader.shutdown()
 
-    @abstractmethod
-    def export(self):
-        pass
+        for metric_exporter in self._metric_exporters:
+            result = result and metric_exporter.shutdown()
 
-    def shutdown(self):
-        # FIXME this will need a Once wrapper
         self._shutdown = True
 
+        if self._atexit_handler is not None:
+            unregister(self._atexit_handler)
+            self._atexit_handler = None
 
-class View:
-    pass
-
+        return result
 
-class ConsoleMetricExporter(MetricExporter):
-    def export(self):
-        pass
+    def get_meter(
+        self,
+        name: str,
+        version: Optional[str] = None,
+        schema_url: Optional[str] = None,
+    ) -> Meter:
 
+        if self._shutdown:
+            _logger.warning(
+                "A shutdown `MeterProvider` can not provide a `Meter`"
+            )
+            return _DefaultMeter(name, version=version, schema_url=schema_url)
 
-class SDKMetricReader(MetricReader):
-    def collect(self):
-        pass
+        return Meter(InstrumentationInfo(name, version, schema_url), self)
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/metric_exporter.py b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/metric_exporter.py
new file mode 100644
index 00000000000..e9c6d6aae07
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/metric_exporter.py
@@ -0,0 +1,17 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+
+class MetricExporter:
+    pass
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/metric_reader.py b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/metric_reader.py
new file mode 100644
index 00000000000..4bbae6501ca
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/metric_reader.py
@@ -0,0 +1,17 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+
+class MetricReader:
+    pass
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/view.py b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/view.py
new file mode 100644
index 00000000000..483fc631f1d
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/view.py
@@ -0,0 +1,18 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+
+class View:
+    def __init__(self, instrument_name=None):
+        pass
diff --git a/opentelemetry-sdk/tests/metrics/test_metrics.py b/opentelemetry-sdk/tests/metrics/test_metrics.py
index b473f4f69fb..4ac3d077151 100644
--- a/opentelemetry-sdk/tests/metrics/test_metrics.py
+++ b/opentelemetry-sdk/tests/metrics/test_metrics.py
@@ -15,14 +15,8 @@
 
 from logging import WARNING
 from unittest import TestCase
-from unittest.mock import Mock
 
-from opentelemetry.sdk._metrics import (
-    ConsoleMetricExporter,
-    MeterProvider,
-    SDKMetricReader,
-    View,
-)
+from opentelemetry.sdk._metrics import MeterProvider
 from opentelemetry.sdk.resources import Resource
 
 
@@ -40,7 +34,7 @@ def test_meter_provider_resource(self):
         self.assertIsInstance(meter_provider_1._resource, Resource)
 
         resource = Resource({"key": "value"})
-        self.assertIs(MeterProvider(resource)._resource, resource)
+        self.assertIs(MeterProvider(resource=resource)._resource, resource)
 
     def test_get_meter(self):
         """
@@ -58,75 +52,6 @@ def test_get_meter(self):
         self.assertEqual(meter._instrumentation_info.version, "version")
         self.assertEqual(meter._instrumentation_info.schema_url, "schema_url")
 
-    def test_register_metric_reader(self):
-        """ "
-        `MeterProvider` provides a way to configure `SDKMetricReader`s.
-        """
-
-        meter_provider = MeterProvider()
-
-        self.assertTrue(hasattr(meter_provider, "register_metric_reader"))
-
-        metric_reader = SDKMetricReader()
-
-        meter_provider.register_metric_reader(metric_reader)
-
-        self.assertTrue(meter_provider._metric_readers, [metric_reader])
-
-    def test_register_metric_exporter(self):
-        """ "
-        `MeterProvider` provides a way to configure `ConsoleMetricExporter`s.
-        """
-
-        meter_provider = MeterProvider()
-
-        self.assertTrue(hasattr(meter_provider, "register_metric_exporter"))
-
-        metric_exporter = ConsoleMetricExporter()
-
-        meter_provider.register_metric_exporter(metric_exporter)
-
-        self.assertTrue(meter_provider._metric_exporters, [metric_exporter])
-
-    def test_register_view(self):
-        """ "
-        `MeterProvider` provides a way to configure `View`s.
-        """
-
-        meter_provider = MeterProvider()
-
-        self.assertTrue(hasattr(meter_provider, "register_view"))
-
-        view = View()
-
-        meter_provider.register_view(view)
-
-        self.assertTrue(meter_provider._views, [view])
-
-    def test_meter_configuration(self):
-        """
-        Any updated configuration is applied to all returned `Meter`s.
-        """
-
-        meter_provider = MeterProvider()
-
-        view_0 = View()
-
-        meter_provider.register_view(view_0)
-
-        meter_0 = meter_provider.get_meter("meter_0")
-        meter_1 = meter_provider.get_meter("meter_1")
-
-        self.assertEqual(meter_0._meter_provider._views, [view_0])
-        self.assertEqual(meter_1._meter_provider._views, [view_0])
-
-        view_1 = View()
-
-        meter_provider.register_view(view_1)
-
-        self.assertEqual(meter_0._meter_provider._views, [view_0, view_1])
-        self.assertEqual(meter_1._meter_provider._views, [view_0, view_1])
-
     def test_shutdown_subsequent_calls(self):
         """
         No subsequent attempts to get a `Meter` are allowed after calling
@@ -141,65 +66,3 @@ def test_shutdown_subsequent_calls(self):
 
         with self.assertLogs(level=WARNING):
             meter_provider.shutdown()
-
-    def test_shutdown_result(self):
-        """
-        `MeterProvider.shutdown` provides a way to let the caller know if it
-        succeeded or failed.
-
-        `MeterProvider.shutdown` is implemented by at least invoking
-        ``shutdown`` on all registered `SDKMetricReader`s and `ConsoleMetricExporter`s.
-        """
-
-        meter_provider = MeterProvider()
-
-        meter_provider.register_metric_reader(
-            Mock(**{"shutdown.return_value": True})
-        )
-        meter_provider.register_metric_exporter(
-            Mock(**{"shutdown.return_value": True})
-        )
-
-        self.assertTrue(meter_provider.shutdown())
-
-        meter_provider = MeterProvider()
-
-        meter_provider.register_metric_reader(
-            Mock(**{"shutdown.return_value": True})
-        )
-        meter_provider.register_metric_exporter(
-            Mock(**{"shutdown.return_value": False})
-        )
-
-        self.assertFalse(meter_provider.shutdown())
-
-    def test_force_flush_result(self):
-        """
-        `MeterProvider.force_flush` provides a way to let the caller know if it
-        succeeded or failed.
-
-        `MeterProvider.force_flush` is implemented by at least invoking
-        ``force_flush`` on all registered `SDKMetricReader`s and `ConsoleMetricExporter`s.
-        """
-
-        meter_provider = MeterProvider()
-
-        meter_provider.register_metric_reader(
-            Mock(**{"force_flush.return_value": True})
-        )
-        meter_provider.register_metric_exporter(
-            Mock(**{"force_flush.return_value": True})
-        )
-
-        self.assertTrue(meter_provider.force_flush())
-
-        meter_provider = MeterProvider()
-
-        meter_provider.register_metric_reader(
-            Mock(**{"force_flush.return_value": True})
-        )
-        meter_provider.register_metric_exporter(
-            Mock(**{"force_flush.return_value": False})
-        )
-
-        self.assertFalse(meter_provider.force_flush())